From 7381caa83d6f094f107a9e339d9aef43fdf8f6e8 Mon Sep 17 00:00:00 2001 From: antonrh Date: Mon, 31 Aug 2026 10:34:22 +0300 Subject: [PATCH] Fix injection in mounted apps, late routes and stacked subscribers `install()` now descends into mounted applications, keeps validating routes added after it runs, and the middleware leaves the container in the ASGI scope so a mounted app can reach it. The FastStream extension wires every handler of a subscriber, not just the first. Installing settings twice names the model and the field that collided, and takes `override=True`. Co-Authored-By: Claude Opus 5 (1M context) --- anydi/ext/fastapi.py | 89 +++++++++++++++--- anydi/ext/faststream.py | 3 +- anydi/ext/pydantic_settings.py | 21 +++-- anydi/ext/starlette/middleware.py | 3 + docs/extensions/fastapi.md | 35 ++++++++ docs/extensions/pydantic_settings.md | 10 +++ tests/ext/fastapi/test_ext.py | 110 ++++++++++++++++++++++- tests/ext/faststream/test_subscribers.py | 31 +++++++ tests/ext/test_pydantic.py | 28 ++++++ 9 files changed, 313 insertions(+), 17 deletions(-) diff --git a/anydi/ext/fastapi.py b/anydi/ext/fastapi.py index 195e896e..4502b87e 100644 --- a/anydi/ext/fastapi.py +++ b/anydi/ext/fastapi.py @@ -8,14 +8,16 @@ message = "This extension needs `fastapi`. Install it with `pip install fastapi`." raise ImportError(message) from None +import functools import inspect -from collections.abc import Iterable, Iterator +from collections.abc import Callable, Iterable, Iterator from typing import Annotated, Any, cast from fastapi import Depends, FastAPI, params, routing as fastapi_routing from fastapi.dependencies.models import Dependant from fastapi.routing import APIRoute, APIWebSocketRoute from starlette.requests import HTTPConnection +from starlette.routing import Mount from anydi import Container, Inject from anydi._marker import Marker, extend_marker @@ -29,8 +31,23 @@ _IncludedRouter: Any = getattr(fastapi_routing, "_IncludedRouter", None) +CONTAINER_SCOPE_KEY = "anydi.container" +"""Where the middleware leaves the container, for apps mounted under others.""" + + def get_container(connection: HTTPConnection) -> Container: - return cast(Container, connection.app.state.container) + container = connection.scope.get(CONTAINER_SCOPE_KEY) + if container is None: + # A mounted application has state of its own, which nobody filled. + container = getattr(connection.app.state, "container", None) + if container is None: + raise RuntimeError( + "No container is installed on this application. Call " + "`anydi.ext.fastapi.install(app, container)`, and add " + "`RequestScopedMiddleware` to the outermost application when it " + "mounts others." + ) + return cast(Container, container) class FastAPIMarker(Marker, params.Depends): @@ -50,21 +67,29 @@ async def _fastapi_dependency( return await container.aresolve(self.dependency_type) -# Configure Inject() and Provide[T] to use FastAPI-specific marker at import time -# This is also called in install() to ensure it's set correctly even if other -# extensions have overwritten it +# Composes with a marker another extension installed, so both keep working. extend_marker(FastAPIMarker) def _iter_routes( routes: Iterable[Any], ) -> Iterator[APIRoute | APIWebSocketRoute]: - """Yield all API routes, descending into routers added via include_router.""" + """Yield all API routes, descending into included routers and mounted apps.""" for route in routes: if isinstance(route, APIRoute | APIWebSocketRoute): yield route elif _IncludedRouter is not None and isinstance(route, _IncludedRouter): yield from _iter_routes(route.original_router.routes) + elif isinstance(route, Mount): + yield from _iter_routes(getattr(route.app, "routes", ())) + + +def _iter_apps(app: Any) -> Iterator[Any]: + """Yield the application and every application mounted under it.""" + yield app + for route in getattr(app, "routes", ()): + if isinstance(route, Mount) and hasattr(route.app, "routes"): + yield from _iter_apps(route.app) def _iter_dependencies(dependant: Dependant) -> Iterator[Dependant]: @@ -104,16 +129,60 @@ def _validate_route_dependencies( marker.set_owner("fastapi") +_REGISTERS_ROUTES = ( + "add_api_route", + "add_api_websocket_route", + "include_router", + "mount", +) + + +def _validate_routes( + app: FastAPI, container: Container, patched: set[tuple[Any, ...]] +) -> None: + """Give every marker of every route its owner.""" + for route in _iter_routes(app.routes): + _validate_route_dependencies(route, container, patched) + + +def _watch_routes(app: FastAPI, patched: set[tuple[Any, ...]]) -> None: + """Validate routes added after `install()`, as they are added.""" + if getattr(app.state, "anydi_watching_routes", False): + return + app.state.anydi_watching_routes = True + router = app.router + + def watching(register: Callable[..., Any]) -> Callable[..., Any]: + @functools.wraps(register) + def wrapper(*args: Any, **kwargs: Any) -> Any: + result = register(*args, **kwargs) + container = getattr(app.state, "container", None) + if container is not None: + # `patched` carries over, so only the new routes are inspected. + _validate_routes(app, container, patched) + return result + + return wrapper + + for name in _REGISTERS_ROUTES: + register = getattr(router, name, None) + if register is not None: + setattr(router, name, watching(register)) + + def install(app: FastAPI, container: Container) -> None: """Install AnyDI into a FastAPI application.""" app.state.container = container + for mounted in _iter_apps(app): + state = getattr(mounted, "state", None) + # A sub-application installed with a container of its own keeps it. + if state is not None and getattr(state, "container", None) is None: + state.container = container # Register websocket scope with request as parent if not already registered if not container.has_scope("websocket"): container.register_scope("websocket", parents=["request"]) - # Validate routes (both HTTP and WebSocket), including routes registered - # through include_router. patched: set[tuple[Any, ...]] = set() - for route in _iter_routes(app.routes): - _validate_route_dependencies(route, container, patched) + _validate_routes(app, container, patched) + _watch_routes(app, patched) diff --git a/anydi/ext/faststream.py b/anydi/ext/faststream.py index b0b86596..06c0f5b9 100644 --- a/anydi/ext/faststream.py +++ b/anydi/ext/faststream.py @@ -77,7 +77,8 @@ async def _faststream_dependency(self, context: ContextRepo) -> Any: def _get_broker_handlers(broker: BrokerUsecase[Any, Any]) -> list[Any]: - return [subscriber.calls[0].handler for subscriber in broker.subscribers] + # A subscriber holds one call per handler registered on it, not just one. + return [call.handler for sub in broker.subscribers for call in sub.calls] def install(broker: BrokerUsecase[Any, Any], container: Container) -> None: diff --git a/anydi/ext/pydantic_settings.py b/anydi/ext/pydantic_settings.py index f9c68d62..27326f02 100644 --- a/anydi/ext/pydantic_settings.py +++ b/anydi/ext/pydantic_settings.py @@ -23,6 +23,7 @@ def install( container: Container, *, prefix: str = "settings.", + override: bool = False, ) -> None: """Install Pydantic settings into an AnyDI container.""" @@ -41,11 +42,21 @@ def _register_settings(_settings: BaseSettings) -> None: else: continue - container.register( - Annotated[origin, f"{prefix}{setting_name}"], - _get_setting_value(getattr(_settings, setting_name)), - scope="singleton", - ) + interface = f"{prefix}{setting_name}" + try: + container.register( + Annotated[origin, interface], + _get_setting_value(getattr(_settings, setting_name)), + scope="singleton", + override=override, + ) + except LookupError: + # Say which field collided, not just which annotation. + raise LookupError( + f"`{settings_cls.__name__}.{setting_name}` is already " + f"registered as `{interface}`. Install it under another " + "`prefix`, or pass `override=True` to replace it." + ) from None if isinstance(settings, BaseSettings): _register_settings(settings) diff --git a/anydi/ext/starlette/middleware.py b/anydi/ext/starlette/middleware.py index b9ff3f12..9318ecfd 100644 --- a/anydi/ext/starlette/middleware.py +++ b/anydi/ext/starlette/middleware.py @@ -32,6 +32,9 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: await self.app(scope, receive, send) return + # A mounted application replaces `scope["app"]` with itself. + scope["anydi.container"] = self.container + async with AsyncExitStack() as stack: # Create request context first (parent scope) request_context = await stack.enter_async_context( diff --git a/docs/extensions/fastapi.md b/docs/extensions/fastapi.md index f9f22452..9e11fcee 100644 --- a/docs/extensions/fastapi.md +++ b/docs/extensions/fastapi.md @@ -52,6 +52,41 @@ async def say_hello( ``` +Routes added after `install` are validated as they are added, so `install` and route registration can go in either order. + + +## Mounted applications + +`install` walks the applications mounted under the one it is given, so the routes of a sub-application resolve from the same container. Add the `RequestScopedMiddleware` to the outer application: + +```python +from fastapi import FastAPI +from starlette.middleware import Middleware + +import anydi.ext.fastapi +from anydi import Container, Provide +from anydi.ext.fastapi import RequestScopedMiddleware + +container = Container() +container.register(HelloService) + +admin = FastAPI() + + +@admin.get("/hello") +async def say_hello(hello_service: Provide[HelloService]) -> str: + return await hello_service.say_hello(name="admin") + + +app = FastAPI(middleware=[Middleware(RequestScopedMiddleware, container=container)]) +app.mount("/admin", admin) + +anydi.ext.fastapi.install(app, container) +``` + +A sub-application that was installed with a container of its own keeps it. + + ## Lifespan support If you need to use `AnyDI` resources in your `FastAPI` application, you can add `AnyDI` startup and shutdown events to the `FastAPI` lifecycle events. diff --git a/docs/extensions/pydantic_settings.md b/docs/extensions/pydantic_settings.md index fa6fdd00..3ecd8015 100644 --- a/docs/extensions/pydantic_settings.md +++ b/docs/extensions/pydantic_settings.md @@ -43,3 +43,13 @@ anydi.ext.pydantic_settings.install(Settings(), container, prefix="my_settings") app_name = container.resolve(Annotated[str, "my_settings.app_name"]) ``` + +## Installing twice + +Installing two settings models that share a field name raises a `LookupError` naming the model and the field. Give one of them its own prefix, or pass `override=True` to replace the values already in the container: + +```python +anydi.ext.pydantic_settings.install(Settings(app_name="Other App"), container, override=True) + +assert container.resolve(Annotated[str, "settings.app_name"]) == "Other App" +``` diff --git a/tests/ext/fastapi/test_ext.py b/tests/ext/fastapi/test_ext.py index b6c124b7..57e22e44 100644 --- a/tests/ext/fastapi/test_ext.py +++ b/tests/ext/fastapi/test_ext.py @@ -2,12 +2,13 @@ from typing import Annotated, Any import pytest -from fastapi import APIRouter, FastAPI, WebSocket +from fastapi import APIRouter, Depends, FastAPI, WebSocket from starlette.middleware import Middleware from starlette.testclient import TestClient import anydi.ext.fastapi from anydi import Container, Inject, Provide +from anydi.ext.fastapi import get_container from anydi.ext.starlette.middleware import RequestScopedMiddleware @@ -129,6 +130,113 @@ def say_hello(message: str = Inject()) -> Any: anydi.ext.fastapi.install(app, container) +def test_install_with_mounted_app() -> None: + """Routes of a mounted application resolve through the outer container.""" + container = Container() + + @container.provider(scope="singleton") + def message() -> str: + return "mounted" + + sub = FastAPI() + + @sub.get("/leaf") + def leaf(message: Provide[str]) -> Any: + return message + + app = FastAPI(middleware=[Middleware(RequestScopedMiddleware, container=container)]) + app.mount("/sub", sub) + + anydi.ext.fastapi.install(app, container) + + client = TestClient(app) + + assert client.get("/sub/leaf").json() == "mounted" + + +def test_install_keeps_container_of_mounted_app() -> None: + """A mounted application installed with its own container keeps it.""" + outer = Container() + inner = Container() + + @inner.provider(scope="singleton") + def message() -> str: + return "inner" + + sub = FastAPI() + anydi.ext.fastapi.install(sub, inner) + + app = FastAPI() + app.mount("/sub", sub) + anydi.ext.fastapi.install(app, outer) + + assert sub.state.container is inner + + +def test_install_validates_routes_added_later() -> None: + """A route added after install is validated as it is added.""" + container = Container() + + @container.provider(scope="singleton") + def message() -> str: + return "late" + + app = FastAPI(middleware=[Middleware(RequestScopedMiddleware, container=container)]) + + anydi.ext.fastapi.install(app, container) + + @app.get("/late") + def late(message: Provide[str]) -> Any: + return message + + router = APIRouter() + + @router.get("/late-router") + def late_router(message: str = Inject()) -> Any: + return message + + app.include_router(router) + + client = TestClient(app) + + assert client.get("/late").json() == "late" + assert client.get("/late-router").json() == "late" + + +def test_install_validates_route_added_later_unknown_annotation() -> None: + """A route added after install fails the same way as one added before.""" + container = Container() + app = FastAPI() + + anydi.ext.fastapi.install(app, container) + + with pytest.raises( + LookupError, + match=( + r"`(.*?).say_hello` has an unknown dependency parameter `message` " + "with an annotation of `str`." + ), + ): + + @app.get("/hello") + def say_hello(message: str = Inject()) -> Any: + return message + + +def test_get_container_without_install() -> None: + """The error names what to call instead of raising an AttributeError.""" + app = FastAPI() + + @app.get("/hello") + def say_hello(container: Annotated[Container, Depends(get_container)]) -> Any: + return str(container) + + client = TestClient(app, raise_server_exceptions=True) + + with pytest.raises(RuntimeError, match="No container is installed"): + client.get("/hello") + + def test_install_registers_websocket_scope() -> None: """Test that websocket scope is automatically registered during install.""" # Create a fresh container without websocket scope diff --git a/tests/ext/faststream/test_subscribers.py b/tests/ext/faststream/test_subscribers.py index ca4f9dee..c9671a73 100644 --- a/tests/ext/faststream/test_subscribers.py +++ b/tests/ext/faststream/test_subscribers.py @@ -80,6 +80,25 @@ async def process_request_provide( ) -> None: _request_context_results.append(f"{message}:{ctx.request_id}") + # One subscriber carrying two handlers, not just the first one + multi = broker.subscriber("email.multi") + + @multi(filter=lambda msg: msg.content_type == "application/json") + async def send_multi_json( + message: EmailMessage, + mail_service: MailService = Inject(), + ) -> None: + result = await mail_service.send_mail(email=message.to, message="json") + _email_results.append((result.email, result.message)) + + @multi + async def send_multi_text( + message: str, + mail_service: MailService = Inject(), + ) -> None: + result = await mail_service.send_mail(email=message, message="text") + _email_results.append((result.email, result.message)) + anydi.ext.faststream.install(broker, container) return broker @@ -125,6 +144,18 @@ async def test_singleton_dependency_with_provide_annotation( assert _email_results[0] == ("admin@example.com", "Alert: System notification") +async def test_every_handler_of_a_subscriber_is_injected( + broker: RedisBroker, +) -> None: + """Test injection into the second handler registered on one subscriber.""" + _email_results.clear() + + async with TestRedisBroker(broker) as br: + await br.publish("user@example.com", channel="email.multi") + + assert _email_results == [("user@example.com", "text")] + + async def test_request_scoped_dependency_with_inject_marker( broker: RedisBroker, ) -> None: diff --git a/tests/ext/test_pydantic.py b/tests/ext/test_pydantic.py index cf4b7172..51a3b02f 100644 --- a/tests/ext/test_pydantic.py +++ b/tests/ext/test_pydantic.py @@ -1,5 +1,6 @@ from typing import Annotated +import pytest from pydantic import computed_field from pydantic_settings import BaseSettings @@ -32,6 +33,33 @@ def test_install_settings() -> None: assert container.resolve(Annotated[str, "settings.param_computed"]) == "computed" +def test_install_twice_reports_the_field() -> None: + container = Container() + + anydi.ext.pydantic_settings.install(Settings(), container) + + with pytest.raises( + LookupError, + match=( + r"`Settings.param_str` is already registered as `settings.param_str`\. " + r"Install it under another `prefix`, or pass `override=True` to " + r"replace it\." + ), + ): + anydi.ext.pydantic_settings.install(Settings(), container) + + +def test_install_twice_with_override() -> None: + container = Container() + + anydi.ext.pydantic_settings.install(Settings(), container) + anydi.ext.pydantic_settings.install( + Settings(param_str="changed"), container, override=True + ) + + assert container.resolve(Annotated[str, "settings.param_str"]) == "changed" + + def test_install_multiple_settings() -> None: container = Container()