Soupape is a dependency injection library for Python, designed to simplify the management of dependencies, keep your codebase clean and promote a modular architecture.
Soupape is what powers Bolinette under the hood, but it requires no framework and can be used in any Python project.
Installation
Soupape has two external dependencies.
- Peritype, which is used for type inspection.
- Hafersack, which allows storing metadata inside objects.
Soupape requires Python 3.12 and above.
$ pip install soupape
Install it through pip or use your favorite package manager.
Typing
Soupape is an opinionated dependency injection library that leverages Python's type hints to automatically resolve and inject dependencies. Any code that runs through Soupape's injectors must be fully type hinted.
Service constructors, for built-in injection, and custom resolvers, must be fully type hinted or Soupape will raise errors at startup or runtime.
Services
Any type that Soupape manages is called a service, whether you let Soupape instantiate it or use a custom resolver to provide it. Services can depend on other services, which will be automatically resolved and injected by Soupape. The dependency chain can be as deep as needed but circular dependencies are not allowed and will raise errors at injection time.
Services must be registered in a ServiceCollection before they
can be resolved by an injector.
from soupape import ServiceCollection
class ServiceA: ...
class ServiceB:
def __init__(self, service_a: ServiceA) -> None:
self.service_a = service_a
services = ServiceCollection()
services.add_singleton(ServiceA)
services.add_scoped(ServiceB)
Service lifetimes
Soupape supports three service lifetimes: singleton, scoped and transient.
-
Singleton services are registered with
add_singleton().- They are instantiated only once per root injector and the same instance is returned on every resolution, even in scoped injectors.
- They are disposed when the root injector is disposed. If their instantiation is triggered in a scoped injector, they will still be disposed when the root injector is disposed.
- They can only depend on transient and singleton services.
-
Scoped services are registered with
add_scoped().- They cannot be instantiated from the root injector. Their instantiation must be triggered from a scoped injector. Doing so from the root injector will raise an error.
- They are instantiated once per injector scope. Every resolution from the same scope and its child scopes will return the same instance.
- They are disposed when the scoped injector they were instantiated from is disposed.
- They can depend on transient, scoped and singleton services.
-
Transient services are registered with
add_transient().- They are instantiated every time they are resolved, whether from the root injector or a scoped injector.
- They are disposed when the injector they were instantiated from is disposed.
- They can only depend on transient and singleton services.
Interfaces
It is possible to register a service using antoher type or interface as its key. This is useful when you want to register a service as an implementation of an interface or an abstract base class.
from soupape import ServiceCollection
class BaseService[T]: ...
class Service(BaseService[int]): ...
services = ServiceCollection()
services.add_singleton(BaseService[int], Service)
services.add_singleton(Service)
The interface is the first parameter. The implementation does not have to be a subclass of the interface, you can register any type. It's the implementation that will be instantiated when the interface is resolved.
In the example above, Service has been registered both as itself
and as an implementation of BaseService[int]. Resolving either
type will return the same singleton instance. In Soupape, instances are always
registered under their concrete type, no matter what interface they were
registered with.
Injectors
Soupape provides two injectors, AsyncInjector and
SyncInjector. The async injector unlocks the full potential of
Soupape and should be used whenever possible. But the sync injector is also
available for projects that do not use async code.
An injector is created with a service collection as its sole argument.
Injectors implement the context manager protocol, and
should be used with the with statement to ensure
proper resource management. The async injector should be used with
async with.
from soupape import ServiceCollection, AsyncInjector, SyncInjector
class ServiceA: ...
class ServiceB:
def __init__(self, service_a: ServiceA) -> None:
self.service_a = service_a
def define_services() -> ServiceCollection:
services = ServiceCollection()
services.add_singleton(ServiceA)
services.add_singleton(ServiceB)
return services
async def main() -> None:
services = define_services()
# Using the async injector
async with AsyncInjector(services) as injector:
service_b = await injector.require(ServiceB)
# Using the sync injector
with SyncInjector(services) as injector:
service_b = injector.require(ServiceB)
Services are resolved recursively by inspecting their constructors and injecting their dependencies automatically. The instances are created immediately, that is why circular dependencies are not allowed and will raise errors at injection time.
If any service is not registered in the service collection, an error will be
raised at injection time, unless the type hint is marked as optional (e.g.,
Optional[Service] or Service | None). In that case,
None will be injected instead.
Scoped Injectors
Scopes allow you to create child injectors that can instantiate scoped services. This feature is useful to manage lifetimes of services that should be created and disposed of together, such as per-request services in web applications, while keeping the singleton services shared across the entire application lifetime, or any other lifetime management strategy.
A scoped injector is created from a parent injector using the
get_scoped_injector() method.
Like the root injector, scoped injectors should be used as context managers to ensure proper resource management. All scoped services will be disposed of when the scope is exited.
from soupape import ServiceCollection, AsyncInjector, SyncInjector
class ServiceA: ...
class ServiceB:
def __init__(self, service_a: ServiceA) -> None:
self.service_a = service_a
def define_services() -> ServiceCollection:
services = ServiceCollection()
services.add_singleton(ServiceA)
services.add_scoped(ServiceB)
return services
async def main() -> None:
services = define_services()
# Using the async injector
async with AsyncInjector(services) as injector:
async with injector.get_scoped_injector() as scoped_injector:
service_b = await scoped_injector.require(ServiceB)
# Using the sync injector
with SyncInjector(services) as injector:
with injector.get_scoped_injector() as scoped_injector:
service_b = scoped_injector.require(ServiceB)
When using multi-level scopes, be careful to which injector you are using to resolve services. Scoped services are unique to their scope and its childs. Resolving a scoped service from a parent injector will not return the same instance as resolving it from a child scoped injector, but a scoped injector can resolve scoped services from its parent scopes, even as dependencies of other services.
Built-in injection
When registering a service in the collection with its type, Soupape will manage the injection of its dependencies automatically, based on the type hints of its constructor. The injector will also execute a series of lifetime management instructions, described bellow.
Context manager services
If a service implements the context manager protocol (i.e., it has
__enter__ and __exit__ methods), or the async
context manager protocol (i.e., it has __aenter__ and
__aexit__ methods), the built-in resolver will automatically use
the appropriate protocol to manage the service's lifetime.
__enter__ or __aenter__ will be called when the
service is instantiated, and __exit__ or
__aexit__ will be called when the injector or scope managing the
service is exited.
SyncInjector will only manage services implementing the context
manager protocol and will ignore the async context manager protocol.
AsyncInjector can manage both protocols, but will prefer the
async context manager protocol and ignore the synchronous one if both are
implemented.
from typing import Self
from soupape import ServiceCollection, AsyncInjector
class AsyncService:
async def __aenter__(self) -> Self:
print("Entering async service")
return self
async def __aexit__(self, exc_type, exc_value, traceback) -> None:
print("Exiting async service")
def define_services() -> ServiceCollection:
services = ServiceCollection()
services.add_singleton(AsyncService)
return services
async def main() -> None:
services = define_services()
async with AsyncInjector(services) as injector:
async_service = await injector.require(AsyncService)
In this example, when the injector is entered, the
AsyncService.__aenter__ method is called, and when the injector
is exited, the AsyncService.__aexit__ method is called.
Post-initialization methods
When using the built-in resolver, you can decorate a method with
@post_init to have it called automatically after the service is
instantiated and all its dependencies are injected. These methods will be
called in the order they are defined in the class, from base classes to
derived classes.
Asynchronous methods will be awaited automatically when using
AsyncInjector, but SyncInjector will raise an error
if a @post_init method is async.
Post-initialization methods are a good place to put initialization logic in an orderly manner, especially if you need to use a dependency that is not required in the other parts of the class. Contrary to context manager that must use the dependencies injected in the constructor, post-initialization methods can use any dependency declared as a parameter.
These methods are called after the context manager methods, if any.
Keep in mind that post-initialization methods are called during instantiation, so the circular dependency restrictions still apply.
from soupape import ServiceCollection, AsyncInjector, post_init
class ServiceA:
def __init__(self) -> None:
self.initialized = False
@post_init
async def _initialize(self) -> None:
print("Initializing ServiceA")
self.initialized = True
class ServiceB:
def __init__(self, service_a: ServiceA) -> None:
self.service_a = service_a
@post_init
def _check_service_a(self) -> None:
print(f"ServiceA initialized: {self.service_a.initialized}")
def define_services() -> ServiceCollection:
services = ServiceCollection()
services.add_singleton(ServiceA)
services.add_singleton(ServiceB)
return services
async def main() -> None:
services = define_services()
async with AsyncInjector(services) as injector:
service_b = await injector.require(ServiceB)
Prefixing methods with an underscore is a common convention to indicate that they are intended for internal use only. Post-initialization methods are not part of the public API of the class and should not be called from outside. Ideally, they should not be called directly at all and should only be used by the injection system.
Custom resolvers
The built-in resolver mechanisms provided by Soupape cover most use cases. However, there might be scenarios where you need to implement a custom resolution function.
Such functions must declare the return type that will be used to resolve the
dependency. The parameters can be anything, as long as they can be resolved by
Soupape. Resolver functions can be asynchronous, but only the
AsyncInjector can use them.
from soupape import AsyncInjector, ServiceCollection
from my_app.services.http import HttpService
from my_app.resources import SomeResource, get_some_resource
class MyService:
def __init__(self, http_service: HttpService, some_resource: SomeResource) -> None:
self.http_service = http_service
self.some_resource = some_resource
async def my_service_resolver(
http_service: HttpService,
) -> MyService:
some_resource = await get_some_resource(http_service)
return MyService(http_service, some_resource)
services = ServiceCollection()
services.add_singleton(HttpService)
services.add_scoped(my_service_resolver)
This example is really basic, but it shows how to create a custom resolver function that depends on other services. The function is registered in the service collection, the return type is used as the service registration type.
Like with the built-in resolver, you can register custom resolvers with an interface. The return type of the function will be used as the instance key and the interface to resolve the service.
from my_app.services.interfaces import IMyService
services.add_scoped(IMyService, my_service_resolver)
Resource management
Post initialization methods will not be called for services created with custom resolvers. And you have to make sure to handle disposal of resources yourself if needed. Context manager services will not automatically work either. You can manually use context managers inside your resolver functions if needed.
from soupape import ServiceCollection
async def my_service_resolver() -> MyService:
async with MyService() as my_service:
return my_service
services = ServiceCollection()
services.add_scoped(my_service_resolver)
Generator resolvers
Antother way to manage resources with custom resolvers is to use generator
functions. These functions use the yield statement to provide the
service instance. After the injector scope is closed, the code after the
yield statement is executed, allowing you to clean up resources.
from soupape import ServiceCollection
async def my_async_service_resolver() -> MyAsyncService:
my_service = MyAsyncService()
try:
yield my_service
finally:
await my_service.dispose()
def my_sync_service_resolver() -> MySyncService:
my_service = MySyncService()
try:
yield my_service
finally:
my_service.dispose()
Function calling
Dependency injection is not just for class constructors; you can also use it to call regular functions with automatic dependency resolution.
from soupape import AsyncInjector, ServiceCollection
class HttpService: ...
async def fetch_data(url: str, http_service: HttpService) -> str:
response = await http_service.get(url)
return response.text
services = ServiceCollection()
services.add_singleton(HttpService)
injector = AsyncInjector(services)
data = await injector.call(fetch_data, positional_args=["https://example.com/api/data"])
In this example, we define an asynchronous function
fetch_data that depends on an HttpService. The first
parameter, url, cannot be resolved by the injector, so we provide
it as a positional argument when calling the function. Such arguments must be
placed before any dependencies and provided in the correct order. The injector
will automatically resolve the http_service parameter and call
the function.
AsyncInjector's call() method is asynchronous, and
awaits the result of your function so that you don't have to write
await await injector.call(my_async_function).
Miscellaneous
Require the injector
You can require the injector itself as a dependency in your services. This is useful when you need to resolve other dependencies manually or avoid circular injection while having a circular dependency in your services.
You can require SyncInjector or AsyncInjector if you
know which one will be used to resolve the service. But you can also require
the Injector protocol, which will work with both injectors, but
you will lose type information about whether the injector is synchronous or
asynchronous.
Require the generic type
If your service is generic, you can require the generic type parameters as dependencies. This allows you to write a generic service that can manipulate its concrete type parameters.
from collections.abc import Sequence
from soupape import ServiceCollection
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from my_app.models import User
async def get_async_session() -> AsyncSession: ...
class Repository[T]:
def __init__(self, session: AsyncSession, entity_type: type[T]) -> None:
self.session = session
self.entity_type = entity_type
async def get_all(self) -> Sequence[T]:
result = await self.session.execute(select(self.entity_type))
return result.scalars().all()
services = ServiceCollection()
services.add_scoped(get_async_session)
services.add_scoped(Repository[User])
async def main():
injector = AsyncInjector(services)
user_repository = await injector.require(Repository[User])
users = await user_repository.get_all()
This a concrete example of a generic repository service that can be used to
fetch all entities of a given type from a database using SQLAlchemy. The
entity_type parameter is automatically resolved by Soupape when
the Repository service is instantiated. type[Any]
triggers a special resolver that's always available and the concrete type is
inferred from the generic type parameter used to register the service.
Registration with Any
You can register generic services using Any as their type
parameter. This allows you to create a generic service that can handle any
type without specifying the exact type parameter at registration time.
In the previous example, we could register the Repository service
using Any as its type parameter. Requiring
Repository[User] would still return a properly typed instance of
the service.
This creates a catch-all registration for the generic service. You can still
register specific type parameters if you want to override the generic behavior
for certain types. If you create several Any registrations for
the same generic service, it is not defined which one will be used when
resolving a specific type parameter if the specific type parameter is not
registered.
If your service has multiple generic type parameters, you can specify some and
use Any for others. When looking for a valid resolver, Soupape
will match the first registration that fits the requested type parameters.
Types are only matched exactly, a super type will not match a subtype.