# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# pyformat: disable

"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""

from decimal import Decimal
import functools
import json
import typing
from typing import Any, Dict, Iterable, List, Mapping, Tuple, Union, get_args
import typing_extensions
from typing_extensions import get_origin

import httpx
from pydantic import ConfigDict, create_model
from pydantic_core import from_json

from ..types.basemodel import BaseModel, Nullable, OptionalNullable, Unset


def serialize_decimal(as_str: bool):
    def serialize(d):
        if d is None:
            return None
        if isinstance(d, Unset):
            return d

        if not isinstance(d, Decimal):
            raise ValueError("Expected Decimal object")

        return str(d) if as_str else float(d)

    return serialize


def validate_decimal(d):
    if d is None:
        return None

    if isinstance(d, (Decimal, Unset)):
        return d

    if not isinstance(d, (str, int, float)):
        raise ValueError("Expected string, int or float")

    return Decimal(str(d))


def serialize_float(as_str: bool):
    def serialize(f):
        if f is None:
            return None
        if isinstance(f, Unset):
            return f

        if not isinstance(f, float):
            raise ValueError("Expected float")

        return str(f) if as_str else f

    return serialize


def validate_float(f):
    if f is None:
        return None

    if isinstance(f, (float, Unset)):
        return f

    if not isinstance(f, str):
        raise ValueError("Expected string")

    return float(f)


def serialize_int(as_str: bool):
    def serialize(i):
        if i is None:
            return None
        if isinstance(i, Unset):
            return i

        if not isinstance(i, int):
            raise ValueError("Expected int")

        return str(i) if as_str else i

    return serialize


def validate_int(b):
    if b is None:
        return None

    if isinstance(b, (int, Unset)):
        return b

    if not isinstance(b, str):
        raise ValueError("Expected string")

    return int(b)


def validate_const(v):
    def validate(c):
        if c is None:
            return None

        if v != c:
            raise ValueError(f"Expected {v}")

        return c

    return validate


def unmarshal_json(raw, typ: Any) -> Any:
    return unmarshal(from_json(raw), typ, coerce_iterables=False)


def unmarshal(val, typ: Any, coerce_iterables: bool = True) -> Any:
    if coerce_iterables:
        val = _coerce_iterables_for_type(val, typ)
    unmarshaller = create_model(
        "Unmarshaller",
        body=(typ, ...),
        __config__=ConfigDict(populate_by_name=True, arbitrary_types_allowed=True),
    )

    m = unmarshaller(body=val)

    # pyright: ignore[reportAttributeAccessIssue]
    return m.body  # type: ignore


_CONSTRUCT_UNVALIDATED_MAX_DEPTH = 64


def construct_unvalidated(value: Any, typ: Any, _depth: int = 0) -> Any:
    try:
        return unmarshal(value, typ, coerce_iterables=True)
    except Exception:
        pass
    return _construct_lenient(value, typ, _depth)


def _construct_lenient(value: Any, typ: Any, depth: int) -> Any:
    if depth > _CONSTRUCT_UNVALIDATED_MAX_DEPTH:
        return value

    typ = _resolve_type_alias(typ)
    origin = get_origin(typ)

    if _is_annotated_type(origin):
        args = get_args(typ)
        return _construct_lenient(value, args[0], depth) if args else value

    if is_union(origin):
        variants = [arg for arg in get_args(typ) if arg is not type(None)]
        for variant in variants:
            try:
                return unmarshal(value, variant, coerce_iterables=True)
            except Exception:
                continue
        last_err = None
        for variant in variants:
            try:
                return _construct_lenient(value, variant, depth)
            except Exception as e:  # noqa: BLE001
                last_err = e
                continue
        if value is None and type(None) in get_args(typ):
            return None
        if last_err is not None:
            raise last_err
        return value

    if _is_list_type(typ):
        if not isinstance(value, list):
            raise TypeError(f"expected list for {typ!r}, got {type(value)!r}")
        item_type = get_args(typ)[0] if get_args(typ) else Any
        return [construct_unvalidated(item, item_type, depth + 1) for item in value]

    if _is_mapping_type(typ):
        if not isinstance(value, Mapping):
            raise TypeError(f"expected mapping for {typ!r}, got {type(value)!r}")
        value_type = get_args(typ)[1] if len(get_args(typ)) > 1 else Any
        return {
            key: construct_unvalidated(item, value_type, depth + 1)
            for key, item in value.items()
        }

    if _is_pydantic_model_type(typ):
        if not isinstance(value, Mapping):
            raise TypeError(f"expected mapping for {typ!r}, got {type(value)!r}")
        return _construct_model_lenient(value, typ, depth)

    return value


def _construct_model_lenient(value: Mapping, typ: Any, depth: int) -> Any:
    fields: Dict[str, Any] = {}
    consumed = set()
    for name, field in typ.model_fields.items():
        wire_key = None
        if field.alias is not None and field.alias in value:
            wire_key = field.alias
        elif name in value:
            wire_key = name

        annotation = field.rebuild_annotation()
        if wire_key is not None:
            consumed.add(wire_key)
            fields[name] = construct_unvalidated(value[wire_key], annotation, depth + 1)
        elif field.is_required():
            fields[name] = _default_for_required(annotation, depth)
        else:
            fields[name] = field.get_default(call_default_factory=True)

    for key, item in value.items():
        if key not in consumed and key not in fields:
            fields[key] = item

    return typ.model_construct(**fields)


def _default_for_required(typ: Any, depth: int) -> Any:
    typ = _resolve_type_alias(typ)
    origin = get_origin(typ)

    if _is_annotated_type(origin):
        args = get_args(typ)
        return _default_for_required(args[0], depth) if args else None

    if is_union(origin):
        if type(None) in get_args(typ):
            return None
        for variant in get_args(typ):
            try:
                return _default_for_required(variant, depth)
            except Exception:  # noqa: BLE001
                continue
        return None

    if depth <= _CONSTRUCT_UNVALIDATED_MAX_DEPTH and _is_pydantic_model_type(typ):
        return _construct_model_lenient({}, typ, depth + 1)

    return None


def marshal_json(val, typ):
    if is_nullable(typ) and val is None:
        return "null"

    marshaller = create_model(
        "Marshaller",
        body=(typ, ...),
        __config__=ConfigDict(populate_by_name=True, arbitrary_types_allowed=True),
    )

    m = marshaller(body=val)

    d = m.model_dump(by_alias=True, mode="json", exclude_none=True)

    if len(d) == 0:
        return ""

    return json.dumps(d[next(iter(d))], separators=(",", ":"))


def is_nullable(field):
    origin = get_origin(field)
    if origin is Nullable or origin is OptionalNullable:
        return True

    if not origin is Union or type(None) not in get_args(field):
        return False

    for arg in get_args(field):
        if get_origin(arg) is Nullable or get_origin(arg) is OptionalNullable:
            return True

    return False


def is_union(obj: object) -> bool:
    """
    Returns True if the given object is a typing.Union or typing_extensions.Union.
    """
    return any(
        obj is typing_obj for typing_obj in _get_typing_objects_by_name_of("Union")
    )


def stream_to_text(stream: httpx.Response) -> str:
    return "".join(stream.iter_text())


async def stream_to_text_async(stream: httpx.Response) -> str:
    return "".join([chunk async for chunk in stream.aiter_text()])


def stream_to_bytes(stream: httpx.Response) -> bytes:
    return stream.content


async def stream_to_bytes_async(stream: httpx.Response) -> bytes:
    return await stream.aread()


def get_pydantic_model(data: Any, typ: Any) -> Any:
    if not _contains_pydantic_model(data):
        return unmarshal(data, typ)

    return _coerce_iterables_for_type(data, typ)


def _coerce_iterables_for_type(data: Any, typ: Any) -> Any:
    if data is None or isinstance(data, (BaseModel, Unset)):
        return data

    typ = _resolve_type_alias(typ)
    origin = get_origin(typ)

    if _is_annotated_type(origin):
        args = get_args(typ)
        return _coerce_iterables_for_type(data, args[0]) if args else data

    if is_union(origin):
        for arg in (arg for arg in get_args(typ) if arg is not type(None)):
            coerced = _coerce_iterables_for_type(data, arg)
            if coerced is not data:
                return coerced
        return data

    if _is_list_type(typ):
        item_type = get_args(typ)[0] if get_args(typ) else Any
        if isinstance(data, (str, bytes, bytearray, Mapping)):
            return data
        if isinstance(data, Iterable):
            return [_coerce_iterables_for_type(item, item_type) for item in data]
        return data

    if _is_mapping_type(typ):
        value_type = get_args(typ)[1] if len(get_args(typ)) > 1 else Any
        if isinstance(data, Mapping):
            return {
                key: _coerce_iterables_for_type(value, value_type)
                for key, value in data.items()
            }
        return data

    if _is_pydantic_model_type(typ) and isinstance(data, Mapping):
        coerced = None
        for field_name, field in typ.model_fields.items():
            field_type = field.annotation
            for key in (field_name, field.alias):
                if key is not None and key in data:
                    value = data[key] if coerced is None else coerced[key]
                    coerced_value = _coerce_iterables_for_type(value, field_type)
                    if coerced_value is not value:
                        if coerced is None:
                            coerced = dict(data)
                        coerced[key] = coerced_value
        return coerced if coerced is not None else data

    return data


def _resolve_type_alias(typ: Any) -> Any:
    return getattr(typ, "__value__", typ)


def _is_annotated_type(origin: Any) -> bool:
    return any(
        origin is typing_obj
        for typing_obj in _get_typing_objects_by_name_of("Annotated")
    )


def _is_list_type(typ: Any) -> bool:
    typ = _resolve_type_alias(typ)
    return typ is list or get_origin(typ) is list


def _is_mapping_type(typ: Any) -> bool:
    typ = _resolve_type_alias(typ)
    origin = get_origin(typ)
    mapping_origin = get_origin(Mapping[Any, Any])
    return typ in (dict, Dict, Mapping) or origin in (dict, Mapping, mapping_origin)


def _is_pydantic_model_type(typ: Any) -> bool:
    return isinstance(typ, type) and issubclass(typ, BaseModel)


def _contains_pydantic_model(data: Any) -> bool:
    if isinstance(data, BaseModel):
        return True
    if isinstance(data, List):
        return any(_contains_pydantic_model(item) for item in data)
    if isinstance(data, Dict):
        return any(_contains_pydantic_model(value) for value in data.values())

    return False


@functools.cache
def _get_typing_objects_by_name_of(name: str) -> Tuple[Any, ...]:
    """
    Get typing objects by name from typing and typing_extensions.
    Reference: https://typing-extensions.readthedocs.io/en/latest/#runtime-use-of-types
    """
    result = tuple(
        getattr(module, name)
        for module in (typing, typing_extensions)
        if hasattr(module, name)
    )
    if not result:
        raise ValueError(
            f"Neither typing nor typing_extensions has an object called {name!r}"
        )
    return result