# Generated by smithplates HTTP codegen. Do not edit.

from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol


@dataclass(frozen=True)
class AuthCredential:
    scheme_id: str
    value: str


class AuthProvider(Protocol):
    def resolve_auth(self, scheme_id: str) -> AuthCredential | None: ...


@dataclass(frozen=True)
class AuthSchemeBinding:
    scheme_id: str
    location: str
    name: str
    prefix: str | None


@dataclass(frozen=True)
class OperationAuthBinding:
    alternatives: tuple[str, ...]
    allows_anonymous: bool


AUTH_SCHEME_BINDINGS: dict[str, AuthSchemeBinding] = {
    "smithy.api#httpBearerAuth": AuthSchemeBinding(
        scheme_id="smithy.api#httpBearerAuth",
        location="header",
        name="Authorization",
        prefix="Bearer",
    ),
    "smithy.api#httpApiKeyAuth": AuthSchemeBinding(
        scheme_id="smithy.api#httpApiKeyAuth",
        location="header",
        name="X-API-Key",
        prefix="ApiKey",
    ),
    "smithplates.codegen.http#httpCookieAuth": AuthSchemeBinding(
        scheme_id="smithplates.codegen.http#httpCookieAuth",
        location="cookie",
        name="session",
        prefix=None,
    ),
}


OPERATION_AUTH_BINDINGS: dict[str, OperationAuthBinding] = {
    "api_key_only": OperationAuthBinding(
        alternatives=("smithy.api#httpApiKeyAuth",),
        allows_anonymous=False,
    ),
    "ApiKeyOnly": OperationAuthBinding(
        alternatives=("smithy.api#httpApiKeyAuth",),
        allows_anonymous=False,
    ),
    "bearer_only": OperationAuthBinding(
        alternatives=("smithy.api#httpBearerAuth",),
        allows_anonymous=False,
    ),
    "BearerOnly": OperationAuthBinding(
        alternatives=("smithy.api#httpBearerAuth",),
        allows_anonymous=False,
    ),
    "cookie_only": OperationAuthBinding(
        alternatives=("smithplates.codegen.http#httpCookieAuth",),
        allows_anonymous=False,
    ),
    "CookieOnly": OperationAuthBinding(
        alternatives=("smithplates.codegen.http#httpCookieAuth",),
        allows_anonymous=False,
    ),
    "optional": OperationAuthBinding(
        alternatives=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
            "smithy.api#noAuth",
        ),
        allows_anonymous=True,
    ),
    "Optional": OperationAuthBinding(
        alternatives=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
            "smithy.api#noAuth",
        ),
        allows_anonymous=True,
    ),
    "public": OperationAuthBinding(
        alternatives=("smithy.api#noAuth",),
        allows_anonymous=True,
    ),
    "Public": OperationAuthBinding(
        alternatives=("smithy.api#noAuth",),
        allows_anonymous=True,
    ),
    "required": OperationAuthBinding(
        alternatives=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
        ),
        allows_anonymous=False,
    ),
    "Required": OperationAuthBinding(
        alternatives=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
        ),
        allows_anonymous=False,
    ),
}


def _set_header(headers: dict[str, str], name: str, value: str) -> None:
    existing_key = next((key for key in headers if key.lower() == name.lower()), None)
    if existing_key is not None and existing_key != name:
        del headers[existing_key]
    headers[name] = value


def apply_operation_auth(
    binding: OperationAuthBinding,
    auth_provider: AuthProvider | None,
    headers: dict[str, str],
    query_params: list[tuple[str, str]],
) -> None:
    allowed_scheme_ids = {scheme_id for scheme_id in binding.alternatives if scheme_id != "smithy.api#noAuth"}
    for scheme_id in binding.alternatives:
        if scheme_id == "smithy.api#noAuth":
            return
        if auth_provider is None:
            continue
        credential = auth_provider.resolve_auth(scheme_id)
        if credential is None:
            continue
        if credential.scheme_id not in allowed_scheme_ids:
            msg = f"Auth provider returned scheme {credential.scheme_id!r}, which is not allowed by the operation"
            raise ValueError(msg)
        if credential.scheme_id != scheme_id:
            msg = f"Auth provider returned scheme {credential.scheme_id!r} while resolving {scheme_id!r}"
            raise ValueError(msg)
        if not credential.value:
            continue

        scheme = AUTH_SCHEME_BINDINGS[credential.scheme_id]
        value = f"{scheme.prefix} {credential.value}" if scheme.prefix else credential.value
        if scheme.location == "header":
            _set_header(headers, scheme.name, value)
        elif scheme.location == "query":
            query_params.append((scheme.name, value))
        elif scheme.location == "cookie":
            if any(ord(character) < 0x21 or ord(character) > 0x7E or character in '\\",;' for character in value):
                raise ValueError("Cookie authentication credential contains an invalid character")
            cookie = f"{scheme.name}={value}"
            existing_cookie_key = next((key for key in headers if key.lower() == "cookie"), None)
            if existing_cookie_key is None:
                headers["Cookie"] = cookie
            else:
                headers[existing_cookie_key] = f"{headers[existing_cookie_key]}; {cookie}"
        else:
            msg = f"Unsupported auth location {scheme.location!r}"
            raise ValueError(msg)
        return

    if not binding.allows_anonymous:
        raise RuntimeError("No usable authentication credential is available for the operation")


@dataclass(frozen=True)
class ResponseVariantBinding:
    status_code: int
    media_type: str | None
    header_bindings: tuple[tuple[str, str], ...]
    static_headers: tuple[tuple[str, str], ...]


@dataclass(frozen=True)
class ClientOperationHttpBinding:
    variants_by_type_name: dict[str, ResponseVariantBinding]
    variants_by_status: dict[int, str]

    def variant_for_status(self, status_code: int) -> ResponseVariantBinding:
        type_name = self.variants_by_status.get(status_code)
        if type_name is None:
            msg = f"No HTTP binding registered for response status {status_code}"
            raise KeyError(msg)
        binding = self.variants_by_type_name.get(type_name)
        if binding is None:
            msg = f"No HTTP binding registered for response type {type_name!r}"
            raise KeyError(msg)
        return binding

    def type_name_for_status(self, status_code: int) -> str:
        type_name = self.variants_by_status.get(status_code)
        if type_name is None:
            msg = f"No HTTP binding registered for response status {status_code}"
            raise KeyError(msg)
        return type_name


OPERATION_HTTP_BINDINGS: dict[str, ClientOperationHttpBinding] = {
    "api_key_only": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "ApiKeyOnly": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "bearer_only": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "BearerOnly": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "cookie_only": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "CookieOnly": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "optional": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "Optional": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "public": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "Public": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "required": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
    "Required": ClientOperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
        variants_by_status={
            200: "AuthOutput",
        },
    ),
}
