21766608368

Committed 06 Feb 2026 09:31PM UTC coverage: 98.205% (-0.05%) from 98.25%

Build # 21766608368

Build Type

Pull #45

github

Committed by

web-flow

Commit Message

Merge 915e63055 into af81bb45c

Pull Request Pull Request #45: Add incremental ServiceContainer

Run Details

232 of 233 branches covered (99.57%)

Branch coverage included in aggregate %.

124 of 124 new or added lines in 8 files covered. (100.0%)

5 existing lines in 3 files now uncovered.

2121 of 2163 relevant lines covered (98.06%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.4

/src/wirio/integrations/_fastapi_dependency_injection.py

import functools
import inspect
from collections.abc import AsyncGenerator, Callable
from contextlib import asynccontextmanager
from contextvars import ContextVar
from inspect import Parameter
from typing import TYPE_CHECKING, Any, final

from fastapi import FastAPI
from fastapi.routing import APIRoute
from starlette.requests import Request
from starlette.routing import BaseRoute, Match
from starlette.types import ASGIApp, Receive, Scope, Send
from starlette.websockets import WebSocket

from wirio._service_lookup._parameter_information import (
    ParameterInformation,
)
from wirio._utils._param_utils import ParamUtils
from wirio.abstractions.service_scope import ServiceScope
from wirio.annotations import FromKeyedServicesInjectable
from wirio.exceptions import CannotResolveServiceFromEndpointError

if TYPE_CHECKING:
    from wirio.service_collection import ServiceCollection
    from wirio.service_provider import ServiceProvider


_current_request: ContextVar[Request | WebSocket] = ContextVar(
    "wirio_starlette_request"
)


@final
class FastApiDependencyInjection:
    @classmethod
    def setup(cls, app: FastAPI, services: "ServiceCollection") -> None:
        app.state.wirio_services = services
        app.add_middleware(_WirioAsgiMiddleware)  # ty: ignore[invalid-argument-type]
        cls._update_lifespan(app)
        cls._inject_routes(app.routes)

    @classmethod
    def _update_lifespan(cls, app: FastAPI) -> None:
        old_lifespan = app.router.lifespan_context

        @asynccontextmanager
        async def new_lifespan(app: FastAPI) -> AsyncGenerator[Any]:
            services: ServiceCollection = app.state.wirio_services

            # if it's already built, we assume it's because the user built it manually before calling setup, so we don't build it again
            async with services.build_service_provider() as service_provider:
                app.state.wirio_service_provider = service_provider

                async with old_lifespan(app) as state:
                    yield state

        app.router.lifespan_context = new_lifespan

    @classmethod
    def _are_annotated_parameters_with_wirio_dependencies(
        cls, target: Callable[..., Any]
    ) -> bool:
        for parameter in inspect.signature(target).parameters.values():
            if (
                parameter.annotation is not None
                and hasattr(parameter.annotation, "__metadata__")
                and hasattr(parameter.annotation.__metadata__[0], "dependency")
                and hasattr(
                    parameter.annotation.__metadata__[0].dependency,
                    "__is_wirio_depends__",
                )
            ):
                return True

        return False

    @classmethod
    def _inject_routes(cls, routes: list[BaseRoute]) -> None:
        for route in routes:
            if not (
                isinstance(route, APIRoute)
                and route.dependant.call is not None
                and inspect.iscoroutinefunction(route.dependant.call)
                and cls._are_annotated_parameters_with_wirio_dependencies(
                    route.dependant.call
                )
            ):
                continue

            route.dependant.call = cls._inject_from_container(route.dependant.call)

    @classmethod
    def _inject_from_container(cls, target: Callable[..., Any]) -> Callable[..., Any]:
        @functools.wraps(target)
        async def _inject_async_target(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
            parameters_to_inject = cls._get_parameters_to_inject(target)
            parameters_to_inject_resolved: dict[str, Any] = {
                injected_parameter_name: await cls._resolve_injected_parameter(
                    parameter_information
                )
                for injected_parameter_name, parameter_information in parameters_to_inject.items()
            }
            return await target(*args, **{**kwargs, **parameters_to_inject_resolved})

        return _inject_async_target

    @classmethod
    def _get_request_container(cls) -> ServiceScope:
        """When inside a request, return the scoped container instance handling the current request.

        This is what we almost always want. It has all the information the app container has in addition
        to data specific to the current request.
        """
        return _current_request.get().state.wirio_service_scope

    @classmethod
    def _get_parameters_to_inject(
        cls, target: Callable[..., Any]
    ) -> dict[str, ParameterInformation]:
        result: dict[str, ParameterInformation] = {}

        for parameter_name, parameter in inspect.signature(target).parameters.items():
            if parameter.annotation is Parameter.empty:
                continue

            injectable_dependency = ParamUtils.get_injectable_dependency(parameter)

            if injectable_dependency is None:
                continue

            parameter_information = ParameterInformation(parameter=parameter)
            result[parameter_name] = parameter_information

        return result

    @classmethod
    async def _resolve_injected_parameter(
        cls, parameter_information: ParameterInformation
    ) -> object | None:
        parameter_service: object | None

        if isinstance(
            parameter_information.injectable_dependency, FromKeyedServicesInjectable
        ):
            parameter_service = await cls._get_request_container().service_provider.get_keyed_service_object(
                parameter_information.injectable_dependency.key,
                parameter_information.parameter_type,
            )
        else:
            parameter_service = (
                await cls._get_request_container().service_provider.get_service_object(
                    parameter_information.parameter_type
                )
            )

        if parameter_service is None:
            if parameter_information.is_optional:
                return None

            raise CannotResolveServiceFromEndpointError(
                parameter_information.parameter_type
            )

        return parameter_service


@final
class _WirioAsgiMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] not in {"http", "websocket"}:
            return await self.app(scope, receive, send)

        if scope["type"] == "http":
            request = Request(scope, receive=receive, send=send)
        else:
            request = WebSocket(scope, receive, send)

        token = _current_request.set(request)

        try:
            is_endpoint_matched = False
            is_async_endpoint = False

            for route in scope["app"].routes:
                if (
                    isinstance(route, APIRoute)
                    and route.matches(scope)[0] == Match.FULL
                ):
                    is_endpoint_matched = True
                    original = inspect.unwrap(route.dependant.call)  # pyright: ignore[reportArgumentType]
                    is_async_endpoint = inspect.iscoroutinefunction(original)
                    break

            if not is_endpoint_matched or not is_async_endpoint:
                return await self.app(scope, receive, send)

            services: ServiceProvider = request.app.state.wirio_service_provider

            async with services.create_scope() as service_scope:
                request.state.wirio_service_scope = service_scope
                await self.app(scope, receive, send)
        finally:
            _current_request.reset(token)

1	import functools	1✔
2	import inspect	1✔
3	from collections.abc import AsyncGenerator, Callable	1✔
4	from contextlib import asynccontextmanager	1✔
5	from contextvars import ContextVar	1✔
6	from inspect import Parameter	1✔
7	from typing import TYPE_CHECKING, Any, final	1✔
8
9	from fastapi import FastAPI	1✔
10	from fastapi.routing import APIRoute	1✔
11	from starlette.requests import Request	1✔
12	from starlette.routing import BaseRoute, Match	1✔
13	from starlette.types import ASGIApp, Receive, Scope, Send	1✔
14	from starlette.websockets import WebSocket	1✔
15
16	from wirio._service_lookup._parameter_information import (	1✔
17	ParameterInformation,
18	)
19	from wirio._utils._param_utils import ParamUtils	1✔
20	from wirio.abstractions.service_scope import ServiceScope	1✔
21	from wirio.annotations import FromKeyedServicesInjectable	1✔
22	from wirio.exceptions import CannotResolveServiceFromEndpointError	1✔
23
24	if TYPE_CHECKING:
25	from wirio.service_collection import ServiceCollection
26	from wirio.service_provider import ServiceProvider
27
28
29	_current_request: ContextVar[Request \| WebSocket] = ContextVar(	1✔
30	"wirio_starlette_request"
31	)
32
33
34	@final	1✔
35	class FastApiDependencyInjection:	1✔
36	@classmethod	1✔
37	def setup(cls, app: FastAPI, services: "ServiceCollection") -> None:	1✔
38	app.state.wirio_services = services	1✔
39	app.add_middleware(_WirioAsgiMiddleware) # ty: ignore[invalid-argument-type]	1✔
40	cls._update_lifespan(app)	1✔
41	cls._inject_routes(app.routes)	1✔
42
43	@classmethod	1✔
44	def _update_lifespan(cls, app: FastAPI) -> None:	1✔
45	old_lifespan = app.router.lifespan_context	1✔
46
47	@asynccontextmanager	1✔
48	async def new_lifespan(app: FastAPI) -> AsyncGenerator[Any]:	1✔
49	services: ServiceCollection = app.state.wirio_services	1✔
50
51	# if it's already built, we assume it's because the user built it manually before calling setup, so we don't build it again
52	async with services.build_service_provider() as service_provider:	1✔
53	app.state.wirio_service_provider = service_provider	1✔
54
55	async with old_lifespan(app) as state:	1✔
56	yield state	1✔
57
58	app.router.lifespan_context = new_lifespan	1✔
59
60	@classmethod	1✔
61	def _are_annotated_parameters_with_wirio_dependencies(	1✔
62	cls, target: Callable[..., Any]
63	) -> bool:
64	for parameter in inspect.signature(target).parameters.values():	1✔
65	if (	1✔
66	parameter.annotation is not None
67	and hasattr(parameter.annotation, "__metadata__")
68	and hasattr(parameter.annotation.__metadata__[0], "dependency")
69	and hasattr(
70	parameter.annotation.__metadata__[0].dependency,
71	"__is_wirio_depends__",
72	)
73	):
74	return True	1✔
75
76	return False	1✔
77
78	@classmethod	1✔
79	def _inject_routes(cls, routes: list[BaseRoute]) -> None:	1✔
80	for route in routes:	1✔
81	if not (	1✔
82	isinstance(route, APIRoute)
83	and route.dependant.call is not None
84	and inspect.iscoroutinefunction(route.dependant.call)
85	and cls._are_annotated_parameters_with_wirio_dependencies(
86	route.dependant.call
87	)
88	):
89	continue	1✔
90
91	route.dependant.call = cls._inject_from_container(route.dependant.call)	1✔
92
93	@classmethod	1✔
94	def _inject_from_container(cls, target: Callable[..., Any]) -> Callable[..., Any]:	1✔
95	@functools.wraps(target)	1✔
96	async def _inject_async_target(args: Any, *kwargs: Any) -> Any: # noqa: ANN401	1✔
97	parameters_to_inject = cls._get_parameters_to_inject(target)	1✔
98	parameters_to_inject_resolved: dict[str, Any] = {	1✔
99	injected_parameter_name: await cls._resolve_injected_parameter(
100	parameter_information
101	)
102	for injected_parameter_name, parameter_information in parameters_to_inject.items()
103	}
104	return await target(args, {kwargs, *parameters_to_inject_resolved})	1✔
105
106	return _inject_async_target	1✔
107
108	@classmethod	1✔
109	def _get_request_container(cls) -> ServiceScope:	1✔
110	"""When inside a request, return the scoped container instance handling the current request.
111
112	This is what we almost always want. It has all the information the app container has in addition
113	to data specific to the current request.
114	"""
115	return _current_request.get().state.wirio_service_scope	1✔
116
117	@classmethod	1✔
118	def _get_parameters_to_inject(	1✔
119	cls, target: Callable[..., Any]
120	) -> dict[str, ParameterInformation]:
121	result: dict[str, ParameterInformation] = {}	1✔
122
123	for parameter_name, parameter in inspect.signature(target).parameters.items():	1✔
124	if parameter.annotation is Parameter.empty:	1✔
UNCOV 125	continue	×
126
127	injectable_dependency = ParamUtils.get_injectable_dependency(parameter)	1✔
128
129	if injectable_dependency is None:	1✔
130	continue	1✔
131
132	parameter_information = ParameterInformation(parameter=parameter)	1✔
133	result[parameter_name] = parameter_information	1✔
134
135	return result	1✔
136
137	@classmethod	1✔
138	async def _resolve_injected_parameter(	1✔
139	cls, parameter_information: ParameterInformation
140	) -> object \| None:
141	parameter_service: object \| None
142
143	if isinstance(	1✔
144	parameter_information.injectable_dependency, FromKeyedServicesInjectable
145	):
146	parameter_service = await cls._get_request_container().service_provider.get_keyed_service_object(	1✔
147	parameter_information.injectable_dependency.key,
148	parameter_information.parameter_type,
149	)
150	else:
151	parameter_service = (	1✔
152	await cls._get_request_container().service_provider.get_service_object(
153	parameter_information.parameter_type
154	)
155	)
156
157	if parameter_service is None:	1✔
158	if parameter_information.is_optional:	1✔
159	return None	1✔
160
161	raise CannotResolveServiceFromEndpointError(	1✔
162	parameter_information.parameter_type
163	)
164
165	return parameter_service	1✔
166
167
168	@final	1✔
169	class _WirioAsgiMiddleware:	1✔
170	def __init__(self, app: ASGIApp) -> None:	1✔
171	self.app = app	1✔
172
173	async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:	1✔
174	if scope["type"] not in {"http", "websocket"}:	1✔
175	return await self.app(scope, receive, send)	1✔
176
177	if scope["type"] == "http":	1✔
178	request = Request(scope, receive=receive, send=send)	1✔
179	else:
UNCOV 180	request = WebSocket(scope, receive, send)	×
181
182	token = _current_request.set(request)	1✔
183
184	try:	1✔
185	is_endpoint_matched = False	1✔
186	is_async_endpoint = False	1✔
187
188	for route in scope["app"].routes:	1✔
189	if (	1✔
190	isinstance(route, APIRoute)
191	and route.matches(scope)[0] == Match.FULL
192	):
193	is_endpoint_matched = True	1✔
194	original = inspect.unwrap(route.dependant.call) # pyright: ignore[reportArgumentType]	1✔
195	is_async_endpoint = inspect.iscoroutinefunction(original)	1✔
196	break	1✔
197
198	if not is_endpoint_matched or not is_async_endpoint:	1✔
199	return await self.app(scope, receive, send)	1✔
200
201	services: ServiceProvider = request.app.state.wirio_service_provider	1✔
202
203	async with services.create_scope() as service_scope:	1✔
204	request.state.wirio_service_scope = service_scope	1✔
205	await self.app(scope, receive, send)	1✔
206	finally:
207	_current_request.reset(token)	1✔

AndreuCodina / wirio / 21766608368

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous