DOCSEARCHbyAlgolia
Submit Docs

that-depends

repository·main·Indexed 42 days ago

https://github.com/modern-python/that-depends

A simple, typed, and async-first dependency injection framework for Python 3.10+. It provides an IOC container with support for dependency scopes, type safety compatible with mypy and pyrefly, and inbuilt support for FastAPI, FastStream, and LiteStar. The framework is designed to be production-ready with zero external dependencies.

Tokens
30.6K
Snippets
113
Records
129
Agent score
53%

What's inside that-depends

  1. Overview of that-depends features

    main

    that-depends is a simple, typed dependency injection (DI) framework for Python 3.10+.

    Key features include:

    • Async-first DI: Built with an Inversion of Control (IOC) container designed for asynchronous workflows.
    • Type Safety: Full coverage by type annotations, compatible with mypy in strict mode and pyrefly.
    • Framework Compatibility: Inbuilt support for FastAPI, FastStream, and LiteStar.
    • Dependency Scopes: Dependency context management using scopes.
    • Testability: Ability to override dependencies for testing purposes.
    • Zero Dependencies: The package itself has no external dependencies.
    • Seamless Injection: Inject dependencies into functions and coroutines without manual wiring.

  3. What is a Resource Provider and when to use it

    main

    A Resource is a specialized provider designed for dependencies that require a managed lifecycle. Unlike a plain Singleton, a Resource ensures that a dependency is resolved exactly once and cached, but also provides a mechanism for teardown (finalization).

    Use a Resource when your dependency needs:

    1. Single creation: The setup logic runs only once.
    2. Single finalization: A teardown or cleanup step is required (e.g., closing a database connection or a file).
    3. Concurrency safety: You need to ensure that multiple threads or coroutines attempting to resolve the resource simultaneously do not trigger multiple creation steps.

  4. Concurrency safety in Singleton providers

    main

    The Singleton provider is designed to be safe for concurrent use in both asynchronous and multi-threaded environments:

    1. Async Concurrency: If multiple coroutines call .resolve() simultaneously, the factory is guaranteed to be called only once. All coroutines will receive the same instance.
    2. Thread Concurrency: If multiple threads call .resolve_sync() simultaneously, the factory is called only once, and all threads receive the same instance.
    import asyncio
import threading

# Async safety: multiple coroutines get the same instance
async def main():
   results = await asyncio.gather(
      MyContainer.singleton.resolve(),
      MyContainer.singleton.resolve(),
   )

# Thread safety: multiple threads get the same instance
def thread_task():
   instance = MyContainer.singleton.resolve_sync()

threads = [threading.Thread(target=thread_task) for _ in range(5)]
for t in threads:
   t.start()

  5. How the Dependency Injection Container works

    main

    A Container in that-depends serves as a central registry for managing providers and defining your dependency graph. While providers can be defined independently, using a container is recommended if you need to utilize context features.

    To use a container, you subclass BaseContainer and define your providers as class attributes. This allows you to build complex dependency graphs where one provider's output is passed as an argument to another.

    from that_depends import BaseContainer, providers

class Container(BaseContainer):
    config = providers.Singleton(Config)
    session = providers.Factory(create_db_session, config=config.db)
    
    user_repository = providers.Factory(
        UserRepository, 
        session=session.cast,
        config.users
    )

  6. Use container aliases for string injection

    main

    When using string-based injection, you can define an alias within your container class. Setting an alias attribute on a BaseContainer subclass replaces the class name for the purposes of string resolution. Once an alias is set, the original class name can no longer be used in Provide["..."] strings.

    class A(BaseContainer):
    alias = "C" # replaces the container name 'A'
    b = providers.Factory(Config)

@inject
def read(val = Provide["C.b.name"]):
    return val

  7. Define providers directly in `BaseContainer` subclasses

    main

    In version 2.*, BaseContainer and its subclasses are no longer dynamic. You can no longer add providers to a container class after it has been defined. All providers must be declared within the class body.

    Incorrect (v1. style):*

    class MyContainer(BaseContainer):
    pass

MyContainer.my_provider = providers.Resource(some_function)

    Correct (v2. style):*

    class MyContainer(BaseContainer):
    my_provider = providers.Resource(some_function)

  8. Concept: How the Selector provider works

    main

    The providers.Selector chooses between different providers based on a key. The key can be provided in three ways:

    1. A Callable: A function or lambda that returns a string.
    2. A Provider attribute: Using the .cast attribute of another provider to access a specific value.
    3. A Fixed string: A literal string used for static selection.

    Once the key is resolved, the Selector returns the single dependency associated with that key from the provided keyword arguments.

  9. How `@inject` and `Provide[...]` work together

    main

    The injection mechanism relies on two components:

    1. Provide[Container.provider]: A marker used in function signatures as a default value. It tells that-depends which provider from which container should be used if no argument is provided.
    2. @inject: A decorator that intercepts the function call. It scans the function's parameters for any default values wrapped in Provide[...], resolves them via the container, and executes the function with the resolved values.

    If @inject is applied to a function that contains no Provide[...] default values, it will issue a warning: Expected injection, but nothing found. Remove @inject decorator.

    @inject
def my_function(param: str = Provide[MyContainer.some_provider]):
    ...

  10. How to send values to an async generator with dependencies

    main

    Because @inject prevents async generators from receiving values via .asend(), you must resolve dependencies manually within the generator body if you need to use the receive = yield value pattern.

    Note that if you use @inject, the receive variable will always be None.

    async def my_generator() -> typing.AsyncGenerator[float, float]:
    # Resolve dependency manually inside the body
    value = await Container.factory.resolve()
    receive = yield value # (1)! This will be None if @inject is used
    yield receive + value

  11. Retrieve factory providers as callables via `.provider` and `.provider_sync`

    main

    When using Factory or AsyncFactory, you can access special properties to obtain standalone callable handles for your resources. These act as no-argument functions that wrap the container's .resolve() or .resolve_sync() methods.

    • .provider: Returns an async callable. You must await it to resolve the resource.
    • .provider_sync: Returns a sync callable. You call it normally to resolve the resource.

    This is useful for passing the creation logic (the callable) around your application without passing the entire container.

    from that_depends import BaseContainer
from that_depends.providers import Factory, AsyncFactory

def build_sync_message() -> str:
    return "Hello from sync provider!"

async def build_async_message() -> str:
    return "Hello from async provider!"

class MyContainer(BaseContainer):
    sync_message = Factory(build_sync_message)
    async_message = AsyncFactory(build_async_message)

# 1. Synchronous usage
msg_sync = MyContainer.sync_message.provider_sync()

# 2. Asynchronous usage
import asyncio
async def main():
    msg_async = await MyContainer.async_message.provider
    print(msg_async)

asyncio.run(main())

  12. Handle read-only `providers.Dict(...)` in version 4.*

    main

    In version 4.*, providers.Dict(...) resolves to a Mapping implemented as a mappingproxy instead of a mutable dict.

    To preserve the ability to mutate the dictionary (e.g., assigning new keys), wrap the resolved result in dict() at the call site for both synchronous and asynchronous resolution.

    # Synchronous resolution
mapping = dict(MyContainer.mapping.resolve_sync())
mapping["extra"] = "value"

# Asynchronous resolution
mapping = dict(await MyContainer.mapping.resolve())
mapping["extra"] = "value"