Dependency injection in BlackSheep¶
The getting started tutorials demonstrate how route and query string parameters can be directly injected into request handlers through function signatures. Additionally, BlackSheep supports the dependency injection of services configured for the application. This page covers:
- An introduction to dependency injection in BlackSheep, with a focus on Rodi.
- Service resolution.
- Service lifetime.
- Options to create services.
- Examples of dependency injection.
- How to use alternatives to Rodi.
Rodi's documentation
Detailed documentation for Rodi can be found at: Rodi.
Introduction¶
The Application object exposes a services property that can be used to configure services. When the function signature of a request handler references a type that is registered as a service, an instance of that type is automatically injected when the request handler is called.
Consider this example:
- Some context is necessary to handle certain web requests (for example, a database connection pool).
- A class that contains this context can be configured in application services before the application starts.
- Request handlers have this context automatically injected.
Demo¶
Starting from a minimal environment as described in the getting started
tutorial, create a foo.py file with the following
contents, inside a domain folder:
Import the new class in server.py, and register the type in app.services
as in this example:
# server.py
from blacksheep import Application, get
from domain.foo import Foo
app = Application()
app.services.add_scoped(Foo) # <-- register Foo type as a service
@get("/")
def home(foo: Foo): # <-- foo is referenced in type annotation
return f"Hello, {foo.foo}!"
An instance of Foo is injected automatically for every web request to "/".
Dependency injection is implemented in a dedicated library: Rodi. Rodi implements dependency injection in an unobtrusive way: it works by inspecting code and doesn't require altering the source code of the types it resolves.
Service resolution¶
Rodi automatically resolves dependency graphs when a resolved type depends on
other types. In the following example, instances of A are automatically
created when resolving Foo because the __init__ method in Foo requires an
instance of A:
# domain/foo.py
class A:
pass
class Foo:
def __init__(self, dependency: A) -> None:
self.dependency = dependency
Both types must be registered in app.services:
# server.py
from blacksheep import Application, get, text
from domain.foo import A, Foo
app = Application()
app.services.add_transient(A)
app.services.add_scoped(Foo)
@get("/")
def home(foo: Foo):
return text(
f"""
A: {id(foo.dependency)}
"""
)
Produces a response like the following at "/":
Using class annotations¶
It is possible to use class properties, like in the example below:
Understanding service lifetimes¶
rodi supports types having one of these lifetimes:
- singleton - instantiated only once.
- transient - services are instantiated every time they are required.
- scoped - instantiated once per web request.
Consider the following example, where a type A is registered as transient,
B as scoped, C as singleton:
# domain/foo.py
class A:
...
class B:
...
class C:
...
class Foo:
def __init__(self, a1: A, a2: A, b1: B, b2: B, c1: C, c2: C) -> None:
self.a1 = a1
self.a2 = a2
self.b1 = b1
self.b2 = b2
self.c1 = c1
self.c2 = c2
# server.py
from blacksheep import Application, get, text
from domain.foo import A, B, C, Foo
app = Application()
app.services.add_transient(A)
app.services.add_scoped(B)
app.services.add_singleton(C)
app.services.add_scoped(Foo)
@get("/")
def home(foo: Foo):
return text(
f"""
A1: {id(foo.a1)}
A2: {id(foo.a2)}
B1: {id(foo.b1)}
B2: {id(foo.b2)}
C1: {id(foo.c1)}
C2: {id(foo.c2)}
"""
)
Produces responses like the following at "/":
- Transient services are created every time they are needed (A).
- Scoped services are created once per web request (B).
- Singleton services are instantiated only once and reused across the application (C).
Options to create services¶
Rodi provides several ways to define and instantiate services.
- registering an exact instance as a singleton
- registering a concrete class by its type
- registering an abstract class and one of its concrete implementations
- registering a service using a factory function
For detailed information on this subject, refer to the Rodi documentation: Registering types.
Singleton example¶
class ServiceSettings:
def __init__(
self,
oauth_application_id: str,
oauth_application_secret: str
):
self.oauth_application_id = oauth_application_id
self.oauth_application_secret = oauth_application_secret
app.services.add_instance(
ServiceSettings("00000000001", os.environ["OAUTH_APP_SECRET"])
)
Registering a concrete class¶
Registering an abstract class¶
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
from blacksheep.server.responses import json, not_found
# domain class and abstract repository defined in a dedicated package for
# domain objects
@dataclass
class Cat:
id: str
name: str
class CatsRepository(ABC):
@abstractmethod
async def get_cat_by_id(self, id: str) -> Optional[Cat]:
pass
# ------------------
# the concrete implementation will be defined in a dedicated package
class PostgreSQLCatsRepository(CatsRepository):
async def get_cat_by_id(self, id: str) -> Optional[Cat]:
# TODO: implement
raise Exception("Not implemented")
# ------------------
# register the abstract class and its concrete implementation when configuring
# the application
app.services.add_scoped(CatsRepository, PostgreSQLCatsRepository)
# a request handler needing the CatsRepository doesn't need to know about
# the exact implementation (e.g. PostgreSQL, SQLite, etc.)
@get("/api/cats/{cat_id}")
async def get_cat(cat_id: str, repo: CatsRepository):
cat = await repo.get_cat_by_id(cat_id)
if cat is None:
return not_found()
return json(cat)
Using a factory function¶
class Something:
def __init__(self, value: str) -> None:
self.value = value
def something_factory(services, activating_type) -> Something:
return Something("Factory Example")
app.services.add_transient_by_factory(something_factory)
Example: implement a request context¶
A good example of a scoped service is one used to assign each web request with a trace id that can be used to identify requests for logging purposes.
from uuid import UUID, uuid4
class OperationContext:
def __init__(self):
self._trace_id = uuid4()
@property
def trace_id(self) -> UUID:
return self._trace_id
Register the OperationContext type as a scoped service, this way it is
instantiated once per web request:
app.services.add_scoped(OperationContext)
@get("/")
def home(context: OperationContext):
return text(f"Request ID: {context.trace_id}")
Services that require asynchronous initialization¶
Services that require asynchronous initialization can be configured using
application events. The recommended way is using the lifespan context
manager, like described in the example below.
Here are the key points describing the logic of using the lifespan decorator:
- Purpose: The
@app.lifespandecorator is used to define asynchronous setup and teardown logic for an application, such as initializing and disposing of resources. - Setup Phase: Code before the
yieldstatement is executed when the application starts. This is typically used to initialize resources (e.g., creating an HTTP client, database connections, or other services). - Resource Registration: During the setup phase, resources can be registered as services in the application's dependency injection container, making them available for injection into request handlers.
- Teardown Phase: Code after the
yieldstatement is executed when the application stops. This is used to clean up or dispose of resources (e.g., closing connections or releasing memory). - Singleton Resource Management: The
@lifespandecorator is particularly useful for managing singleton resources that need to persist for the application's lifetime. - Example Use Case: In the provided example, an
HTTP clientis created and registered as a singleton during the setup phase, and it is disposed of during the teardown phase.
Otherwise, it is possible to use the on_start callback, like in the following
example, to register a service that requires asynchronous initialization:
import asyncio
from blacksheep import Application, get, text
app = Application()
class Example:
def __init__(self, text):
self.text = text
async def configure_something(app: Application):
await asyncio.sleep(0.5) # simulate 500 ms delay
app.services.add_instance(Example("Hello World"))
app.on_start += configure_something
@get("/")
async def home(service: Example):
return service.text
Services that require disposal can be disposed of in the on_stop callback:
async def dispose_example(app: Application):
service = app.service_provider[Example]
await service.dispose()
app.on_stop += dispose_example
The container protocol¶
Since version 2, BlackSheep supports alternatives to rodi for dependency
injection. The services property of the Application class needs to conform
to the following container protocol:
- The
registermethod to register types. - The
resolvemethod to resolve instances of types. - The
__contains__method to describe whether a type is defined inside the container.
class ContainerProtocol:
"""
Generic interface of DI Container that can register and resolve services,
and tell if a type is configured.
"""
def register(self, obj_type: Union[Type, str], *args, **kwargs):
"""Registers a type in the container, with optional arguments."""
def resolve(self, obj_type: Union[Type[T], str], *args, **kwargs) -> T:
"""Activates an instance of the given type, with optional arguments."""
def __contains__(self, item) -> bool:
"""
Returns a value indicating whether a given type is configured in this container.
"""
The following example demonstrates how to use
punq for dependency injection as an
alternative to rodi.
from typing import Type, TypeVar, Union, cast
import punq
from blacksheep import Application
from blacksheep.messages import Request
from blacksheep.server.controllers import Controller, get
T = TypeVar("T")
class Foo:
def __init__(self) -> None:
self.foo = "Foo"
class PunqDI:
"""
BlackSheep DI container implemented with punq
https://github.com/bobthemighty/punq
"""
def __init__(self, container: punq.Container) -> None:
self.container = container
def register(self, obj_type, *args):
self.container.register(obj_type, *args)
def resolve(self, obj_type: Union[Type[T], str], *args) -> T:
return cast(T, self.container.resolve(obj_type))
def __contains__(self, item) -> bool:
return bool(self.container.registrations[item])
container = punq.Container()
container.register(Foo)
app = Application(services=PunqDI(container), show_error_details=True)
@get("/")
def home(foo: Foo): # <-- foo is referenced in type annotation
return f"Hello, {foo.foo}!"
class Settings:
def __init__(self, greetings: str):
self.greetings = greetings
container.register(Settings, instance=Settings("example"))
class Home(Controller):
def __init__(self, settings: Settings):
# controllers are instantiated dynamically at every web request
self.settings = settings
async def on_request(self, request: Request):
print("[*] Received a request!!")
def greet(self):
return self.settings.greetings
@get("/home")
async def index(self):
return self.greet()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="localhost", port=44777, log_level="debug")
It is also possible to configure the dependency injection container using the
settings namespace, like in the following example:
from blacksheep.settings.di import di_settings
def default_container_factory():
return PunqDI(punq.Container())
di_settings.use(default_container_factory)
Dependency injection libraries vary.
Some features might not be supported when using a different kind of container,
because not all libraries for dependency injection implement the notion of
singleton, scoped, and transient (most only implement singleton and
transient).
Last modified on: 2025-04-22 08:29:25