Function bodies 42 total

class GreetingController:
    """Controller for greeting endpoints."""

    def __init__(self, service: GreeterService):
        self.service = service

    @get("/{name}")
    async def say_hello(self, name: str):
        """Greet a user by name."""
        message = self.service.greet(name)
        return {"message": message}

    @get("/{name}/goodbye")
    async def say_goodbye(self, name: str):
        """Say goodbye to a user."""
        message = self.service.farewell(name)
        return {"message": message}

    async def say_hello(self, name: str):
        """Greet a user by name."""
        message = self.service.greet(name)
        return {"message": message}

    async def say_goodbye(self, name: str):
        """Say goodbye to a user."""
        message = self.service.farewell(name)
        return {"message": message}

def create_app() -> FastAPI:
    config = configuration(YamlTreeSource("application.yaml"))

    container = init(
        modules=[
            "myapp.config",
            "myapp.services",
            "myapp.controllers",
        ],
        config=config,
    )

    return container.get(FastAPI)

class GreeterService:
    def __init__(self, config: GreetingConfig):
        self.language = config.default_language

    def greet(self, name: str) -> str:
        greeting = GREETINGS.get(self.language, "Hello")
        return f"{greeting}, {name}!"

    def farewell(self, name: str) -> str:
        return f"Goodbye, {name}!"

    def greet(self, name: str) -> str:
        greeting = GREETINGS.get(self.language, "Hello")
        return f"{greeting}, {name}!"

    def configure_app(self, app: FastAPI) -> None:
        """Apply configuration to the FastAPI application.

        Args:
            app: The FastAPI application instance to configure.
        """
        ...

class FastApiSettings:
    """Type-safe application settings for the FastAPI instance.

    Populated automatically from configuration sources (YAML, env, dict)
    using the ``fastapi`` prefix via pico-ioc's ``@configured`` decorator.
    The ``FastApiAppFactory`` converts these fields into keyword arguments
    for the ``FastAPI()`` constructor.

    Attributes:
        title: API title shown in the OpenAPI docs.
        version: API version string.
        debug: Enable FastAPI debug mode.

    Example:
        .. code-block:: yaml

            # application.yaml
            fastapi:
              title: My API
              version: 2.0.0
              debug: true
    """

    title: str = "Pico-FastAPI App"
    version: str = "1.0.0"
    debug: bool = False

class RouteInfo(TypedDict):
    """Metadata attached to a controller method by a route decorator.

    Attributes:
        method: HTTP method string (``"GET"``, ``"POST"``, etc.) or
            ``"WEBSOCKET"``.
        path: URL path for the route.
        kwargs: Extra keyword arguments forwarded to FastAPI's route
            registration (e.g. ``response_model``, ``status_code``).
    """

    method: str
    path: str
    kwargs: Dict[str, Any]

def controller(
    cls: Optional[Type[Any]] = None, *, scope: str = "request", **kwargs: Any
) -> Callable[[Type[Any]], Type[Any]] | Type[Any]:
    """Mark a class as a pico-fastapi controller with DI and auto-routing.

    The decorated class is registered as a pico-ioc ``@component`` and its
    methods decorated with ``@get``, ``@post``, etc. are automatically
    registered as FastAPI routes at startup.

    Can be used with or without arguments::

        @controller
        class A: ...

        @controller(prefix="/api", tags=["Items"], scope="request")
        class B: ...

    Args:
        cls: The class being decorated (passed implicitly when used without
            parentheses).
        scope: pico-ioc scope for the controller instance.  Typical values
            are ``"request"`` (default) and ``"websocket"``.
        **kwargs: Additional metadata forwarded to the ``APIRouter``
            constructor.  Common keys: ``prefix``, ``tags``,
            ``dependencies``, ``

def _create_route_decorator(method: str, path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Create a decorator that attaches route metadata to a function.

    Args:
        method: HTTP method string (e.g. ``"GET"``) or ``"WEBSOCKET"``.
        path: URL path for the route.
        **kwargs: Extra keyword arguments forwarded to FastAPI's route
            registration.

    Returns:
        A decorator that stores a :class:`RouteInfo` dict on the wrapped
        function under the ``_pico_route_info`` attribute.
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        route_info: RouteInfo = {"method": method, "path": path, "kwargs": kwargs}
        setattr(func, PICO_ROUTE_KEY, cast(RouteInfo, route_info))
        return func

    return decorator

def get(path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Define a GET endpoint on a controller method.

    Args:
        path: URL path for the route (e.g. ``"/"``, ``"/{id}"``).
        **kwargs: Additional FastAPI route parameters such as
            ``response_model``, ``status_code``, ``tags``, ``summary``.

    Returns:
        A decorator that attaches GET route metadata to the method.

    Example:
        .. code-block:: python

            @get("/items", response_model=list[Item])
            async def list_items(self):
                return self.service.get_all()
    """
    return _create_route_decorator("GET", path, **kwargs)

def post(path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Define a POST endpoint on a controller method.

    Args:
        path: URL path for the route.
        **kwargs: Additional FastAPI route parameters such as
            ``response_model``, ``status_code``.

    Returns:
        A decorator that attaches POST route metadata to the method.

    Example:
        .. code-block:: python

            @post("/items", status_code=201)
            async def create_item(self, data: ItemCreate):
                return self.service.create(data)
    """
    return _create_route_decorator("POST", path, **kwargs)

def put(path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Define a PUT endpoint on a controller method.

    Args:
        path: URL path for the route.
        **kwargs: Additional FastAPI route parameters.

    Returns:
        A decorator that attaches PUT route metadata to the method.

    Example:
        .. code-block:: python

            @put("/items/{id}")
            async def update_item(self, id: int, data: ItemUpdate):
                return self.service.update(id, data)
    """
    return _create_route_decorator("PUT", path, **kwargs)

def delete(path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Define a DELETE endpoint on a controller method.

    Args:
        path: URL path for the route.
        **kwargs: Additional FastAPI route parameters.

    Returns:
        A decorator that attaches DELETE route metadata to the method.

    Example:
        .. code-block:: python

            @delete("/items/{id}", status_code=204)
            async def delete_item(self, id: int):
                pass
    """
    return _create_route_decorator("DELETE", path, **kwargs)

def patch(path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Define a PATCH endpoint on a controller method.

    Args:
        path: URL path for the route.
        **kwargs: Additional FastAPI route parameters.

    Returns:
        A decorator that attaches PATCH route metadata to the method.

    Example:
        .. code-block:: python

            @patch("/items/{id}")
            async def patch_item(self, id: int, data: ItemPatch):
                return self.service.patch(id, data)
    """
    return _create_route_decorator("PATCH", path, **kwargs)

def websocket(path: str, **kwargs: Any) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Define a WebSocket endpoint on a controller method.

    The WebSocket parameter is detected by type annotation
    (``ws: WebSocket``), not by argument name.

    Args:
        path: WebSocket route path (e.g. ``"/ws/chat"``).
        **kwargs: Additional FastAPI WebSocket route parameters.

    Returns:
        A decorator that attaches WebSocket route metadata to the method.

    Example:
        .. code-block:: python

            from fastapi import WebSocket

            @controller(scope="websocket")
            class ChatController:
                @websocket("/ws/chat")
                async def chat(self, ws: WebSocket):
                    await ws.accept()
                    data = await ws.receive_text()
                    await ws.send_text(f"Echo: {data}")
    """
    return _create_route_decorator("WEBSOCKET", path, **kwargs)

class PicoFastAPIError(Exception):
    """Base exception for all pico-fastapi errors.

    Catch this at startup boundaries to handle any pico-fastapi failure
    without matching individual subclasses.

    Example:
        .. code-block:: python

            try:
                app = container.get(FastAPI)
            except PicoFastAPIError as exc:
                logger.error("pico-fastapi startup failed: %s", exc)
                raise
    """

class NoControllersFoundError(PicoFastAPIError):
    """Raised when no ``@controller``-decorated classes are found at startup.

    This error is raised by ``register_controllers()`` when the pico-ioc
    container does not contain any classes marked with the ``@controller``
    decorator.

    Causes:
        - Controller modules were not included in ``init(modules=[...])``.
        - Controller classes are missing the ``@controller`` decorator.
        - Import errors prevented controller modules from loading.

    Example:
        .. code-block:: python

            try:
                register_controllers(app, container)
            except NoControllersFoundError:
                logger.warning("No controllers found; API has no endpoints.")
    """

    def __init__(self):
        super().__init__("No controllers were registered. Ensure your controller modules are scanned.")

def _priority_of(obj: Any) -> int:
    """Extract the integer priority from a configurer, defaulting to 0.

    Args:
        obj: An object expected to have a ``priority`` attribute.

    Returns:
        The integer priority, or ``0`` if the attribute is missing or
        cannot be converted.
    """
    try:
        return int(getattr(obj, "priority", 0))
    except Exception:
        return 0

def _normalize_http_result(result: Any) -> Response:
    """Convert a controller return value into a Starlette ``Response``.

    Supports the following return conventions:

    - A ``Response`` instance is returned as-is.
    - A ``(content, status_code)`` or ``(content, status_code, headers)``
      tuple is converted to a ``JSONResponse``.
    - Any other value (including Pydantic models) is serialized as JSON
      with status 200.

    Args:
        result: The value returned by a controller method.

    Returns:
        A Starlette ``Response`` suitable for the ASGI pipeline.
    """
    if isinstance(result, Response):
        return result

    if isinstance(result, tuple) and len(result) in (2, 3):
        content, status = result[0], result[1]
        headers = result[2] if len(result) == 3 else None
        # Convert Pydantic models to dict
        if hasattr(content, "model_dump"):
            content = content.model_dump()
        return JSONResponse(content=content, statu

def _create_http_handler(container: PicoContainer, controller_cls: type, method_name: str, sig: inspect.Signature):
    """Create an async HTTP route handler that resolves the controller via DI.

    The handler fetches a fresh controller instance from the container on
    every request, calls the named method, and normalizes the result into
    a ``Response``.

    Args:
        container: The pico-ioc container.
        controller_cls: The controller class to resolve.
        method_name: The name of the method to invoke.
        sig: The method's ``inspect.Signature`` (used to build the
            wrapper's signature, excluding ``self``).

    Returns:
        An async function suitable for ``APIRouter.add_api_route()``.
    """

    async def http_route_handler(**kwargs):
        controller_instance = await container.aget(controller_cls)
        method_to_call = getattr(controller_instance, method_name)
        res = method_to_call(**kwargs)
        if inspect.isawaitable(res):
  

def _create_websocket_handler(container: PicoContainer, controller_cls: type, method_name: str, sig: inspect.Signature):
    """Create an async WebSocket route handler that resolves the controller via DI.

    The handler detects the WebSocket parameter by its type annotation
    (``WebSocket``), not by name, allowing any parameter name
    (e.g. ``ws``, ``socket``).

    Args:
        container: The pico-ioc container.
        controller_cls: The controller class to resolve.
        method_name: The name of the WebSocket method to invoke.
        sig: The method's ``inspect.Signature``.

    Returns:
        An async function suitable for
        ``APIRouter.add_api_websocket_route()``.
    """
    original_params = list(sig.parameters.values())[1:]
    ws_param_name = None
    new_params = []

    for param in original_params:
        if param.annotation is WebSocket:
            ws_param_name = param.name
        else:
            new_params.append(param)

    if not ws_param_name:

def _find_controller_classes(container: PicoContainer) -> list[type]:
    """Find all classes marked with ``@controller`` in the container.

    Inspects the container's internal locator metadata to discover
    registered controller components.

    Args:
        container: The pico-ioc container to search.

    Returns:
        A list of controller classes found in the container.
    """
    locator = getattr(container, "_locator", None)
    if not locator:
        return []
    return [
        key for key, _ in locator._metadata.items() if isinstance(key, type) and getattr(key, IS_CONTROLLER_ATTR, False)
    ]

def _register_route(router: APIRouter, container: PicoContainer, cls: type, name: str, method, route_info: dict):
    """Register a single route on the router.

    Args:
        router: The ``APIRouter`` to add the route to.
        container: The pico-ioc container for DI resolution.
        cls: The controller class owning the method.
        name: The method name.
        method: The unbound method object.
        route_info: A :class:`RouteInfo` dict with ``method``, ``path``,
            and ``kwargs``.
    """
    sig = inspect.signature(method)
    method_type = route_info["method"]

    if method_type == "WEBSOCKET":
        handler_func = _create_websocket_handler(container, cls, name, sig)
        router.add_api_websocket_route(
            path=route_info["path"],
            endpoint=handler_func,
            **route_info["kwargs"],
        )
    else:
        handler_func = _create_http_handler(container, cls, name, sig)
        router.add_api_route(
            path=rout

def _create_router_for_controller(container: PicoContainer, cls: type) -> APIRouter:
    """Create and configure an ``APIRouter`` for a controller class.

    Reads controller metadata (prefix, tags, dependencies, responses)
    from the class and registers all decorated methods as routes.

    Args:
        container: The pico-ioc container.
        cls: The controller class.

    Returns:
        A configured ``APIRouter`` with all the controller's routes.
    """
    meta = getattr(cls, PICO_CONTROLLER_META, {}) or {}
    router = APIRouter(
        prefix=meta.get("prefix", ""),
        tags=meta.get("tags", None),
        dependencies=meta.get("dependencies", None),
        responses=meta.get("responses", None),
    )

    for name, method in inspect.getmembers(cls, inspect.isfunction):
        route_info = getattr(method, PICO_ROUTE_KEY, None)
        if route_info:
            _register_route(router, container, cls, name, method, route_info)

    return router

def register_controllers(app: FastAPI, container: PicoContainer):
    """Discover and register all ``@controller`` classes with the FastAPI app.

    Iterates over all controller classes found in the container's metadata,
    creates an ``APIRouter`` for each, and includes them in the app.

    Args:
        app: The FastAPI application instance.
        container: The pico-ioc container holding controller registrations.

    Raises:
        NoControllersFoundError: If no ``@controller``-decorated classes
            are found in the container.
    """
    controller_classes = _find_controller_classes(container)

    if not controller_classes:
        raise NoControllersFoundError()

    for cls in controller_classes:
        router = _create_router_for_controller(container, cls)
        app.include_router(router)

def _validate_configurers(configurers: List[Any]) -> List[FastApiConfigurer]:
    """Validate and filter configurers, discarding invalid ones with a warning.

    Each configurer is checked against the ``FastApiConfigurer`` protocol.
    Objects that do not satisfy the protocol are logged at WARNING level
    and excluded.

    Args:
        configurers: A list of candidate configurer objects.

    Returns:
        A filtered list containing only valid ``FastApiConfigurer``
        instances.
    """
    valid = []
    for c in configurers:
        if isinstance(c, FastApiConfigurer) and callable(getattr(c, "configure_app", None)):
            valid.append(c)
        else:
            logger.warning("Discarding invalid configurer %r: does not implement FastApiConfigurer protocol", c)
    return valid

def _split_configurers_by_priority(configurers: List[FastApiConfigurer]) -> tuple[list, list]:
    """Split configurers into inner (>= 0) and outer (< 0) groups by priority.

    Args:
        configurers: A list of validated configurers.

    Returns:
        A ``(inner, outer)`` tuple where *inner* configurers have
        ``priority >= 0`` and *outer* configurers have ``priority < 0``.
        Each group is sorted by ascending priority.
    """
    sorted_conf = sorted(configurers, key=_priority_of)
    inner = [c for c in sorted_conf if _priority_of(c) >= 0]
    outer = [c for c in sorted_conf if _priority_of(c) < 0]
    return inner, outer

def _apply_configurers(app: FastAPI, configurers: List[FastApiConfigurer]) -> None:
    """Apply a list of configurers to the app.

    Args:
        app: The FastAPI application instance.
        configurers: Configurers to apply, in order.
    """
    for configurer in configurers:
        configurer.configure_app(app)

def _create_lifespan_manager(container: PicoContainer):
    """Create the lifespan context manager for graceful shutdown.

    The returned context manager performs async cleanup of all container
    resources and then shuts down the container when the application
    stops.

    Args:
        container: The pico-ioc container to manage.

    Returns:
        An async context manager suitable for ``app.router.lifespan_context``.
    """

    @asynccontextmanager
    async def lifespan_manager(app_instance):
        yield
        await container.cleanup_all_async()
        container.shutdown()

    return lifespan_manager

class PicoLifespanConfigurer:
    """Startup hook that wires FastAPI with pico-ioc at application boot.

    This component is auto-discovered by pico-ioc and runs during
    container configuration.  It performs the following steps in order:

    1. Validates and splits configurers by priority.
    2. Applies *inner* configurers (``priority >= 0``).
    3. Adds ``PicoScopeMiddleware``.
    4. Applies *outer* configurers (``priority < 0``).
    5. Registers all ``@controller`` routes.
    6. Attaches the lifespan manager for graceful shutdown.
    """

    @configure
    def setup_fastapi(
        self,
        container: PicoContainer,
        app: FastAPI,
        configurers: List[FastApiConfigurer],
    ) -> None:
        """Wire the FastAPI app with middleware, controllers, and lifespan.

        Args:
            container: The pico-ioc container.
            app: The FastAPI application instance.
            configurers: All discovered ``FastApiConfigurer`` implementations.
    

    def setup_fastapi(
        self,
        container: PicoContainer,
        app: FastAPI,
        configurers: List[FastApiConfigurer],
    ) -> None:
        """Wire the FastAPI app with middleware, controllers, and lifespan.

        Args:
            container: The pico-ioc container.
            app: The FastAPI application instance.
            configurers: All discovered ``FastApiConfigurer`` implementations.
        """
        valid_configurers = _validate_configurers(configurers)
        inner, outer = _split_configurers_by_priority(valid_configurers)

        _apply_configurers(app, inner)
        app.add_middleware(PicoScopeMiddleware, container=container)
        _apply_configurers(app, outer)

        register_controllers(app, container)
        app.router.lifespan_context = _create_lifespan_manager(container)

class FastApiAppFactory:
    """Factory that creates the ``FastAPI`` application as a singleton.

    Reads :class:`FastApiSettings` (populated from configuration sources)
    and passes its fields as keyword arguments to the ``FastAPI()``
    constructor.  The resulting app is registered in the container with
    ``scope="singleton"``.

    Example:
        .. code-block:: python

            from fastapi import FastAPI
            from pico_boot import init

            container = init(modules=["myapp"])
            app = container.get(FastAPI)  # Created by FastApiAppFactory
    """

    @provides(FastAPI, scope="singleton")
    def create_fastapi_app(
        self,
        settings: FastApiSettings,
    ) -> FastAPI:
        """Create a FastAPI instance from the provided settings.

        Args:
            settings: Application settings (title, version, debug).

        Returns:
            A configured ``FastAPI`` application instance.
        """
        return FastAPI(**datacl

    def create_fastapi_app(
        self,
        settings: FastApiSettings,
    ) -> FastAPI:
        """Create a FastAPI instance from the provided settings.

        Args:
            settings: Application settings (title, version, debug).

        Returns:
            A configured ``FastAPI`` application instance.
        """
        return FastAPI(**dataclasses.asdict(settings))

def _cleanup_scope(container: PicoContainer, scope_name: str, scope_id: str) -> None:
    """Clean up a scope if the container has caches.

    Args:
        container: The pico-ioc container instance.
        scope_name: Name of the scope to clean up (e.g. ``"request"``).
        scope_id: Unique identifier of the scope instance.
    """
    if hasattr(container, "_caches"):
        container._caches.cleanup_scope(scope_name, scope_id)

def _get_or_create_session_id(scope: dict) -> str:
    """Get the existing session ID or create a new one.

    Session IDs are stored in the ASGI scope's ``session`` dict under
    the ``pico_session_id`` key.

    Args:
        scope: The ASGI scope dict.  Must contain a ``"session"`` key
            (provided by Starlette's ``SessionMiddleware``).

    Returns:
        The session identifier string (UUID4).
    """
    session = scope["session"]
    if "pico_session_id" not in session:
        session["pico_session_id"] = str(uuid.uuid4())
    return session["pico_session_id"]

class PicoScopeMiddleware:
    """ASGI middleware that manages pico-ioc request, session, and websocket scopes.

    For every HTTP request, a ``request`` scope is created (and optionally a
    ``session`` scope if ``SessionMiddleware`` is active).  For every
    WebSocket connection, a ``websocket`` scope is created.  Scopes are
    cleaned up in a ``finally`` block to prevent memory leaks.

    This middleware is added automatically by ``PicoLifespanConfigurer``
    and should not be added manually.

    Args:
        app: The next ASGI application in the middleware chain.
        container: The pico-ioc container whose scopes are managed.

    Example:
        .. code-block:: python

            # Automatic -- no manual setup required.
            # PicoLifespanConfigurer adds PicoScopeMiddleware for you.
            container = init(modules=["myapp"])
            app = container.get(FastAPI)
    """

    def __init__(self, app, container: PicoContainer):
        self.app = app
    

    def __init__(self, app, container: PicoContainer):
        self.app = app
        self.container = container

    async def __call__(self, scope, receive, send):
        """Dispatch incoming ASGI connections to the appropriate handler.

        Args:
            scope: ASGI connection scope dict.
            receive: ASGI receive callable.
            send: ASGI send callable.
        """
        with self.container.as_current():
            scope_type = scope["type"]

            if scope_type == "http":
                await self._handle_http(scope, receive, send)
            elif scope_type == "websocket":
                await self._handle_websocket(scope, receive, send)
            else:
                await self.app(scope, receive, send)

    async def _handle_http(self, scope, receive, send):
        """Handle HTTP request with request and optional session scopes.

        Creates a unique ``request`` scope for every HTTP request.  If
        a ``"session"`` key is present in the ASGI scope (provided by
        ``SessionMiddleware``), a ``session`` scope is also created.

        Args:
            scope: ASGI HTTP connection scope.
            receive: ASGI receive callable.
            send: ASGI send callable.
        """
        request_id = str(uuid.uuid4())
        try:
            with self.container.scope("request", request_id):
                if "session" in scope:
                    session_id = _get_or_create_session_id(scope)
                    with self.container.scope("session", session_id):
                        await self.app(scope, receive, send)
                else:
                    await self.app(scope, receive, send)
        finally:
            _cleanup_scope(self.container, "request", requ

    async def _handle_websocket(self, scope, receive, send):
        """Handle WebSocket connection with websocket scope.

        Creates a ``websocket`` scope that lasts for the entire WebSocket
        connection lifetime.  The scope is cleaned up when the connection
        closes.

        Args:
            scope: ASGI WebSocket connection scope.
            receive: ASGI receive callable.
            send: ASGI send callable.
        """
        websocket_id = str(uuid.uuid4())
        try:
            with self.container.scope("websocket", websocket_id):
                await self.app(scope, receive, send)
        finally:
            _cleanup_scope(self.container, "websocket", websocket_id)