Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions tests/test_future_annotations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
from __future__ import annotations
import typing

import pytest

from that_depends import BaseContainer, Provide, providers


if typing.TYPE_CHECKING:

class Unresolvable:
pass


class Service:
pass


class FutureAnnotationsContainer(BaseContainer):
service = providers.Object(Service()).bind(Service)


@FutureAnnotationsContainer.inject
def inject_service_sync(service: Service = Provide()) -> Service:
return service


@FutureAnnotationsContainer.inject
async def inject_service_async(service: Service = Provide()) -> Service:
return service


def test_type_based_injection_resolves_postponed_annotations_sync() -> None:
assert isinstance(inject_service_sync(), Service)


async def test_type_based_injection_resolves_postponed_annotations_async() -> None:
assert isinstance(await inject_service_async(), Service)


def test_type_based_injection_rejects_unresolvable_local_annotation() -> None:
class LocalService:
pass

with pytest.raises(TypeError, match="Cannot resolve annotations for injected function"):

@FutureAnnotationsContainer.inject
def target(service: LocalService = Provide()) -> LocalService: # pragma: no cover
return service


def test_type_based_injection_rejects_non_concrete_annotation() -> None:
with pytest.raises(TypeError, match="Type-based injection for 'service' requires a concrete runtime type"):

@FutureAnnotationsContainer.inject
def target(service: list[str] = Provide()) -> list[str]: # pragma: no cover
return service


def test_direct_provider_injection_does_not_resolve_unrelated_annotations() -> None:
provider = providers.Object(1)

@FutureAnnotationsContainer.inject
def target(value: int = Provide[provider], unrelated: Unresolvable | None = None) -> int:
_ = unrelated
return value

assert target() == 1
26 changes: 24 additions & 2 deletions that_depends/injection.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ class _TypedInjectionParameter(typing.NamedTuple):


class _InjectionPlan(typing.NamedTuple):
signature: inspect.Signature
direct_parameters: tuple[_DirectInjectionParameter, ...]
string_parameters: tuple[_StringInjectionParameter, ...]
typed_parameters: tuple[_TypedInjectionParameter, ...]
Expand Down Expand Up @@ -98,10 +99,21 @@ def close(self) -> None:

@functools.cache
def _build_injection_plan(func: typing.Callable[..., typing.Any]) -> _InjectionPlan:
signature = inspect.signature(func)
parameters = tuple(signature.parameters.items())
direct_parameters: list[_DirectInjectionParameter] = []
string_parameters: list[_StringInjectionParameter] = []
typed_parameters: list[_TypedInjectionParameter] = []
for index, (field_name, param) in enumerate(inspect.signature(func).parameters.items()):
if any(isinstance(param.default, _Provide) for _, param in parameters):
try:
resolved_hints = typing.get_type_hints(func)
except (NameError, TypeError) as exc:
msg = f"Cannot resolve annotations for injected function {func.__qualname__}"
raise TypeError(msg) from exc
else:
resolved_hints = {}

for index, (field_name, param) in enumerate(parameters):
default = param.default
if isinstance(default, StringProviderDefinition):
string_parameters.append(_StringInjectionParameter(index, field_name, default))
Expand All @@ -115,14 +127,24 @@ def _build_injection_plan(func: typing.Callable[..., typing.Any]) -> _InjectionP
)
)
elif isinstance(default, _Provide):
annotation = resolved_hints.get(field_name, param.annotation)
if (
annotation is inspect.Parameter.empty
or annotation is typing.Any
or typing.get_origin(annotation) is not None
or not isinstance(annotation, type)
):
msg = f"Type-based injection for {field_name!r} requires a concrete runtime type"
raise TypeError(msg)
typed_parameters.append(
_TypedInjectionParameter(
index,
field_name,
typing.cast(type[typing.Any], param.annotation),
annotation,
)
)
return _InjectionPlan(
signature=signature,
direct_parameters=tuple(direct_parameters),
string_parameters=tuple(string_parameters),
typed_parameters=tuple(typed_parameters),
Expand Down
Loading