# Generated by smithplates HTTP codegen. Do not edit.

from __future__ import annotations

from dataclasses import dataclass

from fastapi import HTTPException, Request
from generated.example.auth_api.app_services import (
    AuthContext,
    AuthCredential,
    AuthVerifier,
)
from pydantic import BaseModel


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

    def extract(self, request: Request) -> tuple[bool, AuthCredential | None]:
        if self.location == "header":
            values = request.headers.getlist(self.name)
        elif self.location == "query":
            values = request.query_params.getlist(self.name)
        elif self.location == "cookie":
            values = []
            for cookie_header in request.headers.getlist("cookie"):
                for part in cookie_header.split(";"):
                    cookie_name, separator, cookie_value = part.strip().partition("=")
                    if separator and cookie_name == self.name:
                        values.append(cookie_value)
        else:
            raise HTTPException(
                status_code=500,
                detail=f"Unknown auth credential location: {self.location}",
            )

        if not values:
            return False, None
        if len(values) != 1:
            raise HTTPException(status_code=401, detail="Ambiguous authentication credential")
        value = values[0]
        if self.prefix is not None:
            supplied_prefix, separator, credential_value = value.partition(" ")
            if not separator or supplied_prefix.casefold() != self.prefix.casefold() or not credential_value:
                return True, None
            value = credential_value
        elif not value:
            return True, None
        return True, AuthCredential(scheme_id=self.scheme_id, value=value)


@dataclass(frozen=True)
class OperationAuthBinding:
    scheme_ids: 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] = {
    "optional": OperationAuthBinding(
        scheme_ids=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
        ),
        allows_anonymous=True,
    ),
    "Optional": OperationAuthBinding(
        scheme_ids=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
        ),
        allows_anonymous=True,
    ),
    "public": OperationAuthBinding(
        scheme_ids=(),
        allows_anonymous=True,
    ),
    "Public": OperationAuthBinding(
        scheme_ids=(),
        allows_anonymous=True,
    ),
    "required": OperationAuthBinding(
        scheme_ids=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
        ),
        allows_anonymous=False,
    ),
    "Required": OperationAuthBinding(
        scheme_ids=(
            "smithy.api#httpBearerAuth",
            "smithy.api#httpApiKeyAuth",
            "smithplates.codegen.http#httpCookieAuth",
        ),
        allows_anonymous=False,
    ),
}


async def _authenticate_request(
    request: Request,
    binding: OperationAuthBinding,
    verifier: AuthVerifier | None,
) -> AuthContext | None:
    if not binding.scheme_ids:
        if binding.allows_anonymous:
            return None
        raise HTTPException(status_code=500, detail="Operation has no authentication alternatives")

    schemes: list[AuthSchemeBinding] = []
    for scheme_id in binding.scheme_ids:
        scheme = AUTH_SCHEME_BINDINGS.get(scheme_id)
        if scheme is None:
            raise HTTPException(
                status_code=500,
                detail=f"Unknown authentication scheme: {scheme_id}",
            )
        schemes.append(scheme)

    supplied_credentials: list[AuthCredential] = []
    credential_was_supplied = False
    for scheme in schemes:
        supplied, credential = scheme.extract(request)
        credential_was_supplied = credential_was_supplied or supplied
        if credential is not None:
            supplied_credentials.append(credential)

    if verifier is None and (credential_was_supplied or not binding.allows_anonymous):
        raise HTTPException(status_code=500, detail="Authentication verifier is not configured")

    if verifier is not None:
        for credential in supplied_credentials:
            identity = await verifier.verify(credential)
            if identity is not None:
                return AuthContext(
                    scheme_id=credential.scheme_id,
                    identity=identity,
                )

    if credential_was_supplied or not binding.allows_anonymous:
        raise HTTPException(status_code=401, detail="Unauthorized")
    return None


async def authenticate_required_request(
    request: Request,
    binding: OperationAuthBinding,
    verifier: AuthVerifier | None,
) -> AuthContext:
    auth = await _authenticate_request(request, binding, verifier)
    if auth is None:
        raise HTTPException(status_code=500, detail="Required authentication resolved anonymously")
    return auth


async def authenticate_optional_request(
    request: Request,
    binding: OperationAuthBinding,
    verifier: AuthVerifier | None,
) -> AuthContext | None:
    return await _authenticate_request(request, binding, verifier)


@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 OperationHttpBinding:
    variants_by_type_name: dict[str, ResponseVariantBinding]

    def variant_for(self, response: BaseModel | None) -> ResponseVariantBinding:
        if response is None:
            binding = self.variants_by_type_name.get(
                "__empty__",
            )
            if binding is None:
                msg = "Service returned None but operation has no empty-response binding"
                raise TypeError(msg)
            return binding
        type_name = type(response).__name__
        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


OPERATION_HTTP_BINDINGS: dict[str, OperationHttpBinding] = {
    "optional": OperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
    ),
    "Optional": OperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
    ),
    "public": OperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
    ),
    "Public": OperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
    ),
    "required": OperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
    ),
    "Required": OperationHttpBinding(
        variants_by_type_name={
            "AuthOutput": ResponseVariantBinding(
                status_code=200,
                media_type="application/json",
                header_bindings=(),
                static_headers=(),
            ),
        },
    ),
}
