Skip to content

Utils

Dependency Injection

Depends

Simple Dependency injection implementation similar to FastAPI.

Source code in germinate_ai/utils/di.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class Depends:
    """Simple Dependency injection implementation similar to FastAPI."""

    _cache: dict[Callable, typ.Any] = {}

    def __init__(self, dependency: Callable, *args: typ.Any, **kwargs: typ.Any):
        self.dependency = dependency
        self.args = args
        self.kwargs = kwargs

    def __call__(self) -> typ.Any:
        if self.dependency in Depends._cache:
            return Depends._cache[self.dependency]

        result = self.dependency(*self.args, **self.kwargs)
        Depends._cache[self.dependency] = result
        return result

get_io_schemas(func)

Get input and output schemas (Pydantic models) from the function signature.

Source code in germinate_ai/utils/di.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def get_io_schemas(func: typ.Callable) -> tuple[BaseModel, BaseModel]:
    """Get input and output schemas (Pydantic models) from the function signature."""
    # Get input and output schemas from signature/type hints
    signature = inspect.signature(func)
    type_hints = typ.get_type_hints(func)

    # might not take any input/output
    input_schema = BaseModel
    output_schema = BaseModel

    if "input" in type_hints:
        input_type = type_hints["input"]
        if issubclass(input_type, BaseModel):
            input_schema = input_type

    if signature.return_annotation == signature.empty:
        # TODO appropriate error type
        raise TypeError(f"Task {func.__name__} does not specify an output schema")
    if "return" in type_hints:
        output_type = type_hints["return"]
        if issubclass(output_type, BaseModel):
            output_schema = output_type

    return input_schema, output_schema

resolve_dependencies(func, *args, **kwargs)

Returns bound arguments with dependencies resolved.

Source code in germinate_ai/utils/di.py 
56
57
58
59
60
61
62
63
64
65
66
def resolve_dependencies(func: typ.Callable, *args: typ.Any, **kwargs: typ.Any) -> inspect.BoundArguments:
    """Returns bound arguments with dependencies resolved."""
    sig = inspect.signature(func)
    bound = sig.bind(*args, **kwargs)
    bound.apply_defaults()

    for k, arg in bound.arguments.items():
        if type(arg) == Depends:
            bound.arguments[k] = arg()

    return bound

resolve_dependencies_wrapper(func)

Returns a function wrapper that binds func arguments, applies defaults and resolves any dependency injections.

Source code in germinate_ai/utils/di.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def resolve_dependencies_wrapper(func: typ.Callable):
    """Returns a function wrapper that binds func arguments, applies defaults and resolves any dependency injections."""
    if asyncio.iscoroutinefunction(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            bound = resolve_dependencies(func, *args, **kwargs)
            return await func(*bound.args, **bound.kwargs)
        return wrapper

    @wraps(func)
    def wrapper(*args, **kwargs):
        bound = resolve_dependencies(func, *args, **kwargs)
        return func(*bound.args, **bound.kwargs)
    return wrapper

Misc

get_next_tick(tick_interval)

Given a tick interval (in seconds), returns a function that gives the time to sleep until next tick based on the last tick timestamp.

Source code in germinate_ai/utils/helpers.py 
 4
 5
 6
 7
 8
 9
10
11
def get_next_tick(tick_interval: int):
    """Given a tick interval (in seconds), returns a function that gives the time to sleep until next tick based on the last tick timestamp."""

    def _next_tick(last_tick: float):
        time_since = time.time() - last_tick
        return tick_interval - time_since

    return _next_tick