Skip to content

Dependency injection in BlackSheep

The getting started tutorials show how route and query string parameters can be injected directly in request handlers, by function signature. BlackSheep also supports dependency injection of services configured for the application. This page describes:

  • 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.


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


Starting from a minimal environment as described in the getting started tutorial, create a file with the following contents, inside a domain folder:

├── domain
│   ├──
│   └──


class Foo:

    def __init__(self) -> None: = "Foo"

Import the new class in, and register the type in as in this example:

from blacksheep import Application, get

from import Foo

app = Application()  # <-- register Foo type as a service

def home(foo: Foo):  # <-- foo is referenced in type annotation
    return f"Hello, {}!"

An instance of Foo is injected automatically for every web request to "/".

Dependency injection is implemented in a dedicated library from the same author: rodi. rodi implements dependency injection in an unobtrusive way: it works by inspecting __init__ methods and doesn't require altering the source code of classes registered as services. rodi can also resolve dependencies by inspecting class annotations, if an __init__ method is not specified for the class to activate.

Service resolution

rodi automatically resolves graphs of services, when a type that is resolved requires other types. In the following example, instances of A are created automatically when resolving Foo because the __init__ method in Foo requires an instance of A:

class A:
    def __init__(self) -> None:

class Foo:
    def __init__(self, a: A) -> None:
        self.a = a

Note that both types need to be registered in

from blacksheep import Application, get, text

from import A, Foo

app = Application()

def home(foo: Foo):
    return text(
        A: {id(foo.a)}

Produces a response like the following at "/":

        A: 140289521293056

Using class annotations

An alternative to defining __init__ methods is to use class annotations, like in the example below:

class A:

class Foo:
    a: A

Understanding service lifetimes

rodi supports services having one of these lifetimes:

  • singleton - instantiated only once per service provider
  • 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:

class A:
    def __init__(self) -> None:

class B:
    def __init__(self) -> None:

class C:
    def __init__(self) -> None:

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

from blacksheep import Application, get, text

from import A, B, C, Foo

app = Application()

def home(foo: Foo):
    return text(
        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 "/":

Request 1:

        A1: 139976289977296

        A2: 139976289977680

        B1: 139976289977584

        B2: 139976289977584

        C1: 139976289978736

        C2: 139976289978736

Request 2:

        A1: 139976289979888

        A2: 139976289979936

        B1: 139976289979988

        B2: 139976289979988

        C1: 139976289978736

        C2: 139976289978736

Note how:

  • transient services are always instantiated whenever they are activated (A)
  • scoped services are instantiated once per web request (B)
  • a singleton service is activated only once (C)

Options to create services

rodi provides several ways to define and instantiate services.

  1. registering an exact instance as a singleton
  2. registering a concrete class by its type
  3. registering an abstract class and one of its concrete implementations
  4. registering a service using a factory function

Singleton example

class ServiceSettings:
    def __init__(
        oauth_application_id: str,
        oauth_application_secret: str
        self.oauth_application_id = oauth_application_id
        self.oauth_application_secret = oauth_application_secret"00000000001", "APP_SECRET_EXAMPLE"))

Registering a concrete class

class HelloHandler:

    def __init__(self):

    def greetings() -> str:
        return "Hello"

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
class Cat:
    id: str
    name: str

class CatsRepository(ABC):

    async def get_cat_by_id(self, id: str) -> Optional[Cat]:

# ------------------

# 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, PostgreSQLCatsRepository)

# a request handler needing the CatsRepository doesn't need to know about
# the exact implementation (e.g. PostgreSQL, SQLite, etc.)
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")

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()

    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:

def home(context: OperationContext):
    return text(
        Request ID: {context.trace_id}

Services that require asynchronous initialization

Services that require asynchronous initialization can be configured inside on_start callbacks, like in the following example:

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"Hello World"))

app.on_start += configure_something

async def home(service: Example):
    return service.text

Services configured this way are automatically injected in request handlers when a parameter name or type annotation matches a key inside

Services that require disposing of should be disposed of in on_stop callback:

async def dispose_example(app: Application):
    # Note: after the application is started, services are read from
    # app.service_provider:

    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:

  • register method to register types
  • resolve method to resolve instances of types
  • __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 shows how to use punq for dependency injection instead of rodi, and how a transient service can be resolved at "/" and a singleton service resolved at "/home":

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: = "Foo"

class PunqDI:
    BlackSheep DI container implemented with 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()

app = Application(services=PunqDI(container), show_error_details=True)

def home(foo: Foo):  # <-- foo is referenced in type annotation
    return f"Hello, {}!"

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

    async def index(self):
        return self.greet()

if __name__ == "__main__":
    import uvicorn, 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())


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: 2023-12-18 17:52:09