Controllers¶

def create_api_keys_controllers[ID](
    *,
    config: ApiKeysControllerConfig[ID] | None = None,
    **options: Unpack[ApiKeysControllerOptions[ID]],
) -> list[type[Controller]]:
    """Return self-service and admin API-key controller classes.

    Returns:
        Generated self-service and admin controller classes.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either ApiKeysControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = ApiKeysControllerConfig(**options) if config is None else config
    create_rate_limit = settings.rate_limit_config.api_key_create if settings.rate_limit_config else None
    update_rate_limit = settings.rate_limit_config.api_key_update if settings.rate_limit_config else None
    ctx = ApiKeysControllerContext(
        id_parser=settings.id_parser,
        create_before_request=_create_before_request_handler(create_rate_limit),
        create_rate_limit_increment=create_api_key_create_increment(create_rate_limit),
        create_rate_limit_reset=create_api_key_create_reset(create_rate_limit),
        update_before_request=_create_before_request_handler(update_rate_limit),
        update_rate_limit_increment=create_api_key_update_increment(update_rate_limit),
        update_rate_limit_reset=create_api_key_update_reset(update_rate_limit),
        security=settings.security,
        require_step_up_on_create=settings.require_step_up_on_create,
        signing_enabled=settings.signing_enabled,
        totp_stepup_policy=dict(settings.totp_stepup_policy),
    )
    self_controller = define_self_api_keys_controller(ctx)
    admin_controller = define_admin_api_keys_controller(ctx)
    self_controller.path = settings.path
    admin_controller.path = settings.users_path
    return [_mark_litestar_auth_route_handler(self_controller), _mark_litestar_auth_route_handler(admin_controller)]

def create_auth_controller[UP: UserProtocol[Any], ID](
    *,
    config: AuthControllerConfig[UP, ID] | None = None,
    **options: Unpack[AuthControllerOptions[UP, ID]],
) -> type[Controller]:
    """Return a controller subclass bound to the provided backend (DI user manager).

    Args:
        config: Auth controller configuration.
        **options: Individual auth controller settings. Do not combine with
            ``config``.

    Returns:
        Controller subclass with backend-specific login and logout handlers.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either AuthControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = AuthControllerConfig(**options) if config is None else config

    _validate_manual_cookie_auth_contract(
        settings.backend,
        csrf_protection_managed_externally=settings.csrf_protection_managed_externally,
        unsafe_testing=settings.unsafe_testing,
    )
    if settings.totp_pending_secret is not None:
        validate_production_secret(
            settings.totp_pending_secret,
            label="totp_pending_secret",
            unsafe_testing=settings.unsafe_testing,
        )
    ctx = _make_auth_controller_context(
        _AuthControllerSettings(
            backend=settings.backend,
            rate_limit_config=settings.rate_limit_config,
            enable_refresh=settings.enable_refresh,
            requires_verification=settings.requires_verification,
            login_identifier=settings.login_identifier,
            totp_pending_secret=settings.totp_pending_secret,
            totp_pending_lifetime=settings.totp_pending_lifetime,
            totp_pending_require_client_binding=settings.totp_pending_require_client_binding,
        ),
    )
    base_cls = _define_auth_controller_class_di(ctx, security=settings.security)
    generated_controller: type[Controller] = (
        _define_refresh_auth_controller_class_di(base_cls, ctx) if ctx.refresh_strategy is not None else base_cls
    )
    generated_controller.__name__ = f"{_build_controller_name(settings.backend.name)}AuthController"
    generated_controller.__qualname__ = generated_controller.__name__
    generated_controller.path = settings.path
    return _mark_litestar_auth_route_handler(generated_controller)

def create_oauth_associate_controller[UP: UserProtocol[Any], ID](
    *,
    config: OAuthAssociateControllerConfig[UP, ID] | None = None,
    **options: Unpack[OAuthAssociateControllerOptions[UP, ID]],
) -> type[Controller]:
    """Return a controller for linking an OAuth account to the authenticated user.

    Both /authorize and /callback are protected by is_authenticated. Callback
    validates account state, then upserts the OAuth account for request.user and
    does not create new users. The generated flow enforces RFC 7636 PKCE S256:
    manual clients must accept ``code_challenge`` / ``code_challenge_method`` on
    authorization and ``code_verifier`` on token exchange.

    Provide either user_manager (for direct use) or user_manager_dependency_key
    (for plugin use with a request-scoped dependency). ``user_manager_dependency_key``
    must be a valid non-keyword Python identifier because Litestar injects
    dependencies by matching keys to callback parameter names.
    ``redirect_base_url`` must use a non-loopback ``https://`` origin; the
    manual controller API does not expose a debug or testing override for
    insecure callback origins.

    Returns:
        Generated controller class mounted under the provider-specific path.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either OAuthAssociateControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = OAuthAssociateControllerConfig(**options) if config is None else config

    return _create_oauth_associate_controller(
        _OAuthAssociateControllerSettings(
            provider_name=settings.provider_name,
            user_manager=settings.user_manager,
            user_manager_dependency_key=settings.user_manager_dependency_key,
            oauth_client=settings.oauth_client,
            redirect_base_url=settings.redirect_base_url,
            oauth_flow_cookie_secret=settings.oauth_flow_cookie_secret,
            path=settings.path,
            cookie_secure=settings.cookie_secure,
            security=settings.security,
            totp_stepup_policy=settings.totp_stepup_policy or None,
        ),
    )

def create_oauth_controller[UP: UserProtocol[Any], ID](
    *,
    config: OAuthControllerConfig[UP, ID] | None = None,
    **options: Unpack[OAuthControllerOptions[UP, ID]],
) -> type[Controller]:
    """Return a controller subclass bound to one OAuth provider.

    The authorize endpoint uses only server-configured ``oauth_scopes``. Runtime
    scope-query overrides are rejected. ``redirect_base_url`` must use a
    non-loopback ``https://`` origin; the manual controller API does not expose
    a debug or testing override for insecure callback origins. The generated
    flow enforces RFC 7636 PKCE S256: manual clients must accept
    ``code_challenge`` / ``code_challenge_method`` on authorization and
    ``code_verifier`` on token exchange.

    Returns:
        Generated controller class mounted under the provider-specific path.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either OAuthControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = OAuthControllerConfig(**options) if config is None else config

    return _create_login_oauth_controller(
        _OAuthLoginControllerSettings(
            provider_name=settings.provider_name,
            backend=settings.backend,
            user_manager=settings.user_manager,
            oauth_client_adapter=_build_oauth_client_adapter(oauth_client=settings.oauth_client),
            redirect_base_url=settings.redirect_base_url,
            oauth_flow_cookie_secret=settings.oauth_flow_cookie_secret,
            path=settings.path,
            cookie_secure=settings.cookie_secure,
            oauth_scopes=settings.oauth_scopes,
            associate_by_email=settings.associate_by_email,
            trust_provider_email_verified=settings.trust_provider_email_verified,
        ),
    )

def create_register_controller[UP: RegisterControllerUserProtocol[Any], ID](
    *,
    config: RegisterControllerConfig | None = None,
    **options: Unpack[RegisterControllerOptions],
) -> type[Controller]:
    """Return a controller subclass for enumeration-resistant public registration.

    Domain failures collapse to HTTP 400 with ``ErrorCode.REGISTER_FAILED`` and
    successful/domain-failed attempts are padded to the configured minimum
    duration.

    Returns:
        Controller subclass exposing the registration endpoint.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either RegisterControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = RegisterControllerConfig(**options) if config is None else config

    _require_msgspec_struct(settings.user_read_schema, parameter_name="user_read_schema")
    _require_msgspec_struct(
        settings.user_create_schema,
        parameter_name="user_create_schema",
        require_forbid_unknown_fields=True,
    )
    register_rate_limit = settings.rate_limit_config.register if settings.rate_limit_config else None
    return _create_register_controller_type(
        _RegisterControllerSettings(
            register_rate_limit=register_rate_limit,
            register_before_request=_create_before_request_handler(register_rate_limit),
            path=settings.path,
            user_read_schema=settings.user_read_schema,
            user_create_schema=settings.user_create_schema,
            minimum_response_seconds=_validate_register_minimum_response_seconds(
                settings.register_minimum_response_seconds,
            ),
            unsafe_testing=settings.unsafe_testing,
        ),
    )

def create_reset_password_controller[UP: ResetPasswordControllerUserProtocol[Any], ID](
    *,
    rate_limit_config: AuthRateLimitConfig | None = None,
    path: str = "/auth",
    user_read_schema: type[msgspec.Struct] = UserRead,
    unsafe_testing: bool = False,
) -> type[Controller]:
    """Return a controller subclass that resolves the user manager via Litestar DI.

    Args:
        rate_limit_config: Optional auth-endpoint rate-limiter configuration.
        path: Base route prefix for the generated controller.
        user_read_schema: Custom msgspec struct used for public reset-password responses.
        unsafe_testing: Explicit test-only override that allows response
            schemas with sensitive fields for isolated fixtures.

    Returns:
        Controller subclass exposing reset-password endpoints.
    """
    _require_msgspec_struct(user_read_schema, parameter_name="user_read_schema")
    forgot_password_rate_limit = rate_limit_config.forgot_password if rate_limit_config else None
    reset_password_rate_limit = rate_limit_config.reset_password if rate_limit_config else None
    reset_password_rate_limit_increment, reset_password_rate_limit_reset = _create_rate_limit_handlers(
        reset_password_rate_limit,
    )
    reset_cls = _define_reset_password_controller_class(
        _ResetPasswordControllerContext(
            user_read_schema=user_read_schema,
            unsafe_testing=unsafe_testing,
            forgot_password_rate_limit=forgot_password_rate_limit,
            forgot_password_before_request=_create_before_request_handler(forgot_password_rate_limit),
            reset_password_before_request=_create_before_request_handler(reset_password_rate_limit),
            reset_password_increment=reset_password_rate_limit_increment,
            reset_password_reset=reset_password_rate_limit_reset,
        ),
    )
    _configure_request_body_handler(cast("Any", reset_cls).reset_password, schema=ResetPassword)
    reset_cls.path = path
    return _mark_litestar_auth_route_handler(reset_cls)

def create_session_devices_controller[UP: UserProtocol[Any], ID](
    config: SessionDevicesControllerConfig[UP, ID] | None = None,
    **options: Unpack[SessionDevicesControllerOptions[UP, ID]],
) -> type[Controller]:
    """Return a controller exposing authenticated refresh-session management routes.

    Raises:
        TypeError: If both ``config`` and keyword options are supplied.
    """
    if config is not None and options:
        msg = "Pass either SessionDevicesControllerConfig or keyword options, not both."
        raise TypeError(msg)
    settings = SessionDevicesControllerConfig(**options) if config is None else config
    generated_controller = _define_session_devices_controller_class(
        _build_static_context(settings.backend),
        security=settings.security,
    )
    generated_controller.__name__ = f"{_build_controller_name(settings.backend.name)}SessionDevicesController"
    generated_controller.__qualname__ = generated_controller.__name__
    generated_controller.path = settings.path.rstrip("/") or "/"
    return _mark_litestar_auth_route_handler(generated_controller)

def create_totp_controller[UP: UserProtocol[Any], ID](
    **options: Unpack[TotpControllerOptions[UP, ID]],
) -> type[Controller]:
    """Return a controller with TOTP management and login-completion endpoints.

    Returns:
        Controller subclass with TOTP management endpoints.
    """
    settings, path, security = _resolve_totp_controller_factory_settings(options)
    return _finalize_totp_controller(settings, path=path, security=security)

def create_users_controller[UP: UsersControllerUserProtocol[Any], ID](
    *,
    config: UsersControllerConfig[ID] | None = None,
    **options: Unpack[UsersControllerOptions[ID]],
) -> type[Controller]:
    """Return the users controller subclass wired for Litestar DI.

    Returns:
        Controller subclass exposing self-service profile, reverified password rotation,
        and admin user endpoints.

    Raises:
        ValueError: If ``config`` and keyword options are combined.
    """
    if config is not None and options:
        msg = "Pass either UsersControllerConfig or keyword options, not both."
        raise ValueError(msg)
    settings = UsersControllerConfig(**options) if config is None else config

    _require_msgspec_struct(settings.user_read_schema, parameter_name="user_read_schema")
    _require_msgspec_struct(
        settings.user_update_schema,
        parameter_name="user_update_schema",
        require_forbid_unknown_fields=True,
    )
    _require_msgspec_struct(
        settings.admin_user_update_schema,
        parameter_name="admin_user_update_schema",
        require_forbid_unknown_fields=True,
    )
    change_password_rate_limit = settings.rate_limit_config.change_password if settings.rate_limit_config else None
    (
        change_password_before_request,
        change_password_rate_limit_increment,
        change_password_rate_limit_reset,
    ) = _create_change_password_rate_limit_handlers(change_password_rate_limit)
    ctx = _UsersControllerContext(
        id_parser=settings.id_parser,
        user_read_schema_type=settings.user_read_schema,
        user_update_schema_type=settings.user_update_schema,
        admin_user_update_schema_type=settings.admin_user_update_schema,
        users_page_schema_type=_create_users_page_schema_type(),
        hard_delete=settings.hard_delete,
        default_limit=settings.default_limit,
        max_limit=settings.max_limit,
        change_password_before_request=change_password_before_request,
        change_password_rate_limit_increment=change_password_rate_limit_increment,
        change_password_rate_limit_reset=change_password_rate_limit_reset,
        unsafe_testing=settings.unsafe_testing,
        totp_stepup_policy=dict(settings.totp_stepup_policy),
    )
    controller_cls = _define_users_controller_class_di(ctx)
    controller_cls.path = settings.path
    if settings.security is not None:
        controller_cls.security = settings.security
    return _mark_litestar_auth_route_handler(controller_cls)

def create_verify_controller[UP: VerifyControllerUserProtocol[Any], ID](  # noqa: PLR0913
    *,
    rate_limit_config: AuthRateLimitConfig | None = None,
    path: str = "/auth",
    user_read_schema: type[msgspec.Struct] = UserRead,
    verify_minimum_response_seconds: float = DEFAULT_MINIMUM_RESPONSE_SECONDS,
    request_verify_minimum_response_seconds: float = DEFAULT_MINIMUM_RESPONSE_SECONDS,
    unsafe_testing: bool = False,
) -> type[Controller]:
    """Return a controller subclass that resolves the user manager via Litestar DI.

    Args:
        rate_limit_config: Optional auth-endpoint rate-limiter configuration. When provided,
            ``request_verify_token`` requests are subject to rate limiting.
        path: Base route prefix for the generated controller.
        user_read_schema: Custom msgspec struct used for public verification responses.
        verify_minimum_response_seconds: Minimum wall-clock duration for verify responses.
        request_verify_minimum_response_seconds: Minimum wall-clock duration for request-verify-token responses.
        unsafe_testing: Explicit test-only override that allows response
            schemas with sensitive fields for isolated fixtures.

    Returns:
        Controller subclass exposing verification-related endpoints.
    """
    _require_msgspec_struct(user_read_schema, parameter_name="user_read_schema")
    verify_rate_limit = rate_limit_config.verify_token if rate_limit_config else None
    request_verify_rate_limit = rate_limit_config.request_verify_token if rate_limit_config else None
    verify_rate_limit_increment, verify_rate_limit_reset = _create_rate_limit_handlers(verify_rate_limit)
    verify_cls = _define_verify_controller_class(
        _VerifyControllerContext(
            user_read_schema=user_read_schema,
            unsafe_testing=unsafe_testing,
            verify_minimum_response_seconds=_validate_minimum_response_seconds(
                verify_minimum_response_seconds,
                field_name="verify_minimum_response_seconds",
            ),
            request_verify_minimum_response_seconds=_validate_minimum_response_seconds(
                request_verify_minimum_response_seconds,
                field_name="request_verify_minimum_response_seconds",
            ),
            verify_before_request=_create_before_request_handler(verify_rate_limit),
            request_verify_before_request=_create_before_request_handler(request_verify_rate_limit),
            verify_increment=verify_rate_limit_increment,
            verify_reset=verify_rate_limit_reset,
            request_verify_rate_limit=request_verify_rate_limit,
        ),
    )
    verify_cls.path = path
    return _mark_litestar_auth_route_handler(verify_cls)