20831478298

Committed 08 Jan 2026 08:55PM UTC coverage: 95.789% (+0.09%) from 95.695%

Build # 20831478298

Build Type

Pull #19

github

Committed by

web-flow

Commit Message

Merge a3f97bf2f into 265b2fc46

Pull Request Pull Request #19: Not infere with non-annotated FastAPI parameters

Run Details

119 of 120 branches covered (99.17%)

Branch coverage included in aggregate %.

12 of 14 new or added lines in 2 files covered. (85.71%)

4 existing lines in 1 file now uncovered.

1064 of 1115 relevant lines covered (95.43%)

0.95 hits per line

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

95.49

/src/aspy_dependency_injection/_integrations/_fastapi_dependency_injection.py

import functools
import inspect
from contextlib import asynccontextmanager
from contextvars import ContextVar
from inspect import Parameter
from typing import TYPE_CHECKING, Any, final

from fastapi.routing import APIRoute
from starlette.requests import Request
from starlette.routing import Match
from starlette.websockets import WebSocket

from aspy_dependency_injection._service_lookup._parameter_information import (
    ParameterInformation,
)
from aspy_dependency_injection.injectable import Injectable

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator, Callable, Sequence

    from fastapi import FastAPI
    from starlette.routing import BaseRoute
    from starlette.types import ASGIApp, Receive, Scope, Send

    from aspy_dependency_injection.service_collection import ServiceCollection
    from aspy_dependency_injection.service_provider import (
        ServiceProvider,
        ServiceScope,
    )

current_request: ContextVar[Request | WebSocket] = ContextVar("aspy_starlette_request")


@final
class FastApiDependencyInjection:
    @classmethod
    def setup(cls, app: FastAPI, services: ServiceCollection) -> None:
        service_provider = services.build_service_provider()
        app.state.aspy_service_provider = service_provider
        app.add_middleware(_AspyAsgiMiddleware)
        cls._update_lifespan(app, service_provider)
        cls._inject_routes(app.routes)

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

        @asynccontextmanager
        async def new_lifespan(app: FastAPI) -> AsyncGenerator[Any]:
            await service_provider.__aenter__()

            async with old_lifespan(app) as state:
                yield state

            await service_provider.__aexit__(None, None, None)

        app.router.lifespan_context = new_lifespan

    @classmethod
    def _are_annotated_parameters_with_aspy_dependencies(
        cls, target: Callable[..., Any]
    ) -> bool:
        for parameter in inspect.signature(target).parameters.values():
            if parameter.annotation is not None and isinstance(
                parameter.annotation, Injectable
            ):
                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 not cls._are_annotated_parameters_with_aspy_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, returns the scoped container instance handling the current request.

        This is what you 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.aspy_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 = cls._get_injectable_dependency(parameter)

            if injectable_dependency is None:
                continue

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

        return result

    @classmethod
    def _get_injectable_dependency(cls, parameter: Parameter) -> Injectable | None:
        if not hasattr(parameter.annotation, "__metadata__"):
            return None

        metadata: Sequence[Any] = parameter.annotation.__metadata__

        for metadata_item in metadata:
            if hasattr(metadata_item, "dependency"):
                dependency = metadata_item.dependency()  # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType, reportAttributeAccessIssue]

                if isinstance(dependency, Injectable):
                    return dependency

        return None

    @classmethod
    async def _resolve_injected_parameter(
        cls, parameter_information: ParameterInformation
    ) -> object | None:
        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

            error_message = f"Unable to resolve service for type '{parameter_information.parameter_type}' while attempting to invoke endpoint"
            raise RuntimeError(error_message)

        return parameter_service


@final
class _AspyAsgiMiddleware:
    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_async_endpoint = False

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

                    if is_async_endpoint:
                        break

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

            service_provider: ServiceProvider = request.app.state.aspy_service_provider

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

AndreuCodina / aspy-dependency-injection / 20831478298

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