Skip to content

Inject, Factory and Injects

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.

Example

@get(dependencies={"value": Inject(lambda: 13)})
def myview(value: Injects()):
    return {"value": value}
Source code in ravyn/injector.py
137
138
139
140
141
142
143
144
145
146
147
148
149
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__"