aspy-dependency-injection 0.6.0

Powerful Dependency Injection with Python

Features

• Use it everywhere: Use dependency injection in web servers, background tasks, console applications, Jupyter notebooks, tests, etc.
• Lifetimes: Singleton (same instance per application), Scoped (same instance per HTTP request scope) and Transient (different instance per resolution).
• FastAPI integration out of the box, and pluggable to any web framework.
• Automatic resolution and disposal: Automatically resolve constructor parameters and manage async and non-async context managers. It's no longer our concern to know how to create or dispose services.
• Clear design inspired by one of the most used and battle-tested DI libraries, adding async-native support, important features and good defaults.
• Centralized configuration: Register all services in one place using a clean syntax, and without decorators.
• ty and Pyright strict compliant.

📦 Installation

uv add aspy-dependency-injection

✨ Quickstart with FastAPI

Inject services into async endpoints using Annotated[..., FromServices()].

class EmailService:
    pass


class UserService:
    def __init__(self, email_service: EmailService) -> None:
        self.email_service = email_service


app = FastAPI()

@app.post("/users")
async def create_user(user_service: Annotated[UserService, FromServices()]) -> None:
    ...

services = ServiceCollection()
services.add_transient(EmailService)
services.add_transient(UserService)
services.configure_fastapi(app)

✨ Quickstart without FastAPI

Define services and create a service provider.

class EmailService:
    pass


class UserService:
    def __init__(self, email_service: EmailService) -> None:
        self.email_service = email_service


services = ServiceCollection()
services.add_transient(EmailService)
services.add_transient(UserService)

async with services.build_service_provider() as service_provider:
    user_service = await service_provider.get_required_service(UserService)

If we want a scope per operation (e.g., per HTTP request or message from a queue), we can create a scope from the service provider:

async with service_provider.create_scope() as service_scope:
    user_service = await service_scope.get_required_service(UserService)

🔄 Lifetimes

• Transient: A new instance is created every time the service is requested. Examples: Services without state, workflows, repositories, service clients...
• Singleton: The same instance is used every time the service is requested. Examples: Settings (pydantic-settings), machine learning models, database connection pools, caches.
• Scoped: A new instance is created for each new scope, but the same instance is returned within the same scope. Examples: Database clients, unit of work.

🏭 Factories

Sometimes, we need to use a factory function to create a service. For example, we have settings (a connection string, database name, etc.) stored using the package pydantic-settings and we want to provide them to a service DatabaseClient to access a database.

class ApplicationSettings(BaseSettings):
    database_connection_string: str


class DatabaseClient:
    def __init__(self, connection_string: str) -> None:
        pass

In a real DatabaseClient implementation, we must use a sync or async context manager, i.e., we instance it with:

async with DatabaseClient(database_connection_string) as client:
    ...

And, if we want to reuse it, we create a factory function with yield:

async def create_database_client(application_settings: ApplicationSettings) -> AsyncGenerator[DatabaseClient]:
    async with DatabaseClient(application_settings.database_connection_string) as database_client:
        yield database_client

With that factory, we have to provide manually a singleton of ApplicationSettings, and to know if DatabaseClient implements a sync or async context manager, or neither. Apart from that, if we need a singleton or scoped instance of DatabaseClient, it's very complex to manage the disposal of the instance.

Then, why don't just return it? With this package, we just have this:

def inject_database_client(application_settings: ApplicationSettings) -> DatabaseClient:
    return DatabaseClient(
        connection_string=application_settings.database_connection_string
    )

services.add_transient(inject_database_client)

The factories can take as parameters other services registered. In this case, inject_database_client takes ApplicationSettings as a parameter, and the dependency injection mechanism will resolve it automatically.

🧪 Simplified testing

We can create a fixture in conftest.py that provides a ServiceProvider instance:

@pytest.fixture
async def service_provider() -> AsyncGenerator[ServiceProvider]:
    async with services.build_service_provider() as service_provider:
        yield service_provider

And then we can inject it into our tests and resolve the services.

async def test_create_user(service_provider: ServiceProvider) -> None:
    user_service = await service_provider.get_required_service(UserService)

📝 Interfaces & abstract classes

We can register a service by specifying both the service type (interface / abstract class) and the implementation type (concrete class). This is useful when we want to inject services using abstractions.

class NotificationService(ABC):
    @abstractmethod
    async def send_notification(self, user_id: str, message: str) -> None:
        ...


class EmailService(NotificationService):
    @override
    async def send_notification(self, user_id: str, message: str) -> None:
        pass


class UserService:
    def __init__(self, notification_service: NotificationService) -> None:
        self.notification_service = notification_service

    async def create_user(self, email: str) -> None:
        user = self.create_user(email)
        await self.notification_service.send_notification(user.id, "Welcome to our service!")


services.add_transient(NotificationService, EmailService)

📝 Keyed services

We can register a service by specifying both the service type and a key. This is useful when we want to resolve services using abstractions and an explicit key.

class NotificationService(ABC):
    @abstractmethod
    async def send_notification(self, user_id: str, message: str) -> None:
        ...


class EmailService(NotificationService):
    @override
    async def send_notification(self, user_id: str, message: str) -> None:
        pass


class PushNotificationService(NotificationService):
    @override
    async def send_notification(self, user_id: str, message: str) -> None:
        pass


class UserService:
    def __init__(
        self,
        notification_service: Annotated[NotificationService, FromKeyedServices("email"),
    ) -> None:
        self.notification_service = notification_service

    async def create_user(self, email: str) -> None:
        user = self.create_user(email)
        await self.notification_service.send_notification(user.id, "Welcome to our service!")


services.add_keyed_transient("email", NotificationService, EmailService)
services.add_keyed_transient("push", NotificationService, PushNotificationService)

📝 Auto-activated services

We can register a service as auto-activated. This is useful when we want to ensure our FastAPI application doesn't start to serve requests until certain services are fully initialized (e.g., machine learning models, database connection pools and caches).

services.add_auto_activated_singleton(MachineLearningModel)

📚 Documentation

For more information, check out the documentation.