Перейти к содержанию

Inject, Factory and Injects

Warning

The current page still doesn't have a translation for this language.

But you can help translating it: Contributing.

Ravyn dependency injection system is actually pretty simple and can be checked in the official dependency injection section for more details.

from ravyn import Inject, Injects, Factory

ravyn.Inject

Inject(dependency, use_cache=False, **kwargs)

Bases: ArbitraryHashableBaseModel

Ravyn's dependency injector built on Lilya's dependency system internally. Keeps full backward compatibility with the old behavior and interface.

Source code in ravyn/injector.py
79
80
81
82
83
84
85
def __init__(self, dependency: "AnyCallable", use_cache: bool = False, **kwargs: Any):
    super().__init__(**kwargs)
    self.dependency = dependency
    self.use_cache = use_cache
    self.signature_model: Optional[Type["SignatureModel"]] = None
    self._depends = _Depends(dependency, use_cache=use_cache)
    self.value: Any = Void

model_config class-attribute instance-attribute

model_config = ConfigDict(
    extra="allow", arbitrary_types_allowed=True
)

dependency instance-attribute

dependency = dependency

use_cache instance-attribute

use_cache = use_cache

signature_model instance-attribute

signature_model = None

_depends instance-attribute

_depends = Depends(dependency, use_cache=use_cache)

value instance-attribute

value = Void

__call__ async

__call__(**kwargs)
Source code in ravyn/injector.py
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def __call__(self, **kwargs: dict[str, Any]) -> Any:
    if self.use_cache and self.value is not Void:
        return self.value

    try:
        if is_async_callable(self.dependency):
            value = await self.dependency(**kwargs)
        else:
            value = self.dependency(**kwargs)
    except TypeError as e:
        raise ImproperlyConfigured(str(e)) from e

    if self.use_cache:
        self.value = value

    return value

__eq__

__eq__(other)
Source code in ravyn/injector.py
104
105
106
107
108
109
110
def __eq__(self, other: Any) -> bool:
    return other is self or (
        isinstance(other, self.__class__)
        and other.dependency == self.dependency
        and other.use_cache == self.use_cache
        and getattr(other, "value", Void) == getattr(self, "value", Void)
    )

__hash__

__hash__()
Source code in ravyn/injector.py
112
113
114
115
116
117
118
119
120
def __hash__(self) -> int:
    values: dict[str, Any] = {}
    for key, value in self.__dict__.items():
        values[key] = None
        if isinstance(value, (list, set)):
            values[key] = tuple(value)
        else:
            values[key] = value
    return hash((type(self),) + tuple(values))
    - "!^model_config"
    - "!^__hash__"
    - "!^__call__"
    - "!^__eq__"

ravyn.Injects

Injects(
    default=Undefined,
    skip_validation=False,
    allow_none=True,
)

Bases: FieldInfo

Creates a FieldInfo class with extra parameters. This is used for dependencies and to inject them.

Note: Intentionally extends Pydantic's @final FieldInfo to enable dependency injection. The DI system relies on isinstance(param, FieldInfo) to detect injection points. Composition would break this detection, so we extend FieldInfo despite its @final marker.

Example

@get(dependencies={"value": Inject(lambda: 13)})
def myview(value: Injects()):
    return {"value": value}
Source code in ravyn/injector.py
141
142
143
144
145
146
147
148
149
150
151
152
153
def __init__(
    self,
    default: Any = Undefined,
    skip_validation: bool = False,
    allow_none: bool = True,
) -> None:
    self.allow_none = allow_none
    self.extra: dict[str, Any] = {
        IS_DEPENDENCY: True,
        SKIP_VALIDATION: skip_validation,
        "allow_none": self.allow_none,
    }
    super().__init__(default=default, json_schema_extra=self.extra)

allow_none instance-attribute

allow_none = allow_none

extra instance-attribute

extra = {
    IS_DEPENDENCY: True,
    SKIP_VALIDATION: skip_validation,
    "allow_none": allow_none,
}
    - "!^model_config"
    - "!^__hash__"
    - "!^__call__"
    - "!^__eq__"

ravyn.Factory

Factory(provides, *args, **kwargs)

A dependency injection factory that supports both positional and keyword arguments.

The provider can be passed as either: - A direct callable - A string reference to be dynamically imported

Example Usage

dependencies = { "user": Factory(UserDAO, db_session=session, cache=cache) }

Source code in ravyn/injector.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, provides: Union["AnyCallable", str], *args: Any, **kwargs: Any) -> None:
    """
    A dependency injection factory that supports both positional and keyword arguments.

    The provider can be passed as either:
    - A direct callable
    - A string reference to be dynamically imported

    Example Usage:
        dependencies = {
            "user": Factory(UserDAO, db_session=session, cache=cache)
        }
    """
    self.__args: tuple[Any, ...] = args
    self.__kwargs: dict[str, Any] = kwargs
    self.is_nested: bool = False

    if isinstance(provides, str):
        self.provides, self.is_nested = load_provider(provides)
    else:
        self.provides = provides

__args instance-attribute

__args = args

__kwargs instance-attribute

__kwargs = kwargs

is_nested instance-attribute

is_nested = False

provides instance-attribute

provides = provides

cls property

cls

Return the provided class or function.

set_args

set_args(*args, **kwargs)

Set or update arguments dynamically.

Source code in ravyn/injector.py
43
44
45
46
def set_args(self, *args: Any, **kwargs: Any) -> None:
    """Set or update arguments dynamically."""
    self.__args = args
    self.__kwargs = kwargs

__call__ async

__call__()

Instantiates the provided class/function, handling both sync and async cases.

Supports
  • Nested imports (e.g., MyClass.func, MyClass.SubClass.func)
  • Both sync and async callables
  • Positional and keyword arguments
Example

Factory(UserDAO, db_session=session)

Source code in ravyn/injector.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def __call__(self) -> Any:
    """
    Instantiates the provided class/function, handling both sync and async cases.

    Supports:
        - Nested imports (e.g., MyClass.func, MyClass.SubClass.func)
        - Both sync and async callables
        - Positional and keyword arguments

    Example:
        Factory(UserDAO, db_session=session)
    """
    if self.is_nested:
        self.provides = self.provides()

    if is_async_callable(self.provides):
        return await self.provides(*self.__args, **self.__kwargs)
    return self.provides(*self.__args, **self.__kwargs)
    - "!^model_config"
    - "!^__hash__"
    - "!^__call__"
    - "!^__eq__"