Service container

The service container is the central piece of any Expanse application. It serves as a global registry where services can be retrieved and/or registered, and is what makes automatic dependency injection possible.

An Expanse application acts as a container, so most of your interactions with the container will be done through the application instance. However, this document will use the container directly to showcase its features.

Registering services

If you want to introduce additional services to your Expanse application, you will need to register them to the container. In a typical Expanse application this is done in service providers, but you can choose to do it manually if needed.

Registering simple services

Registering a service is done with the register() method of the container which accepts a class or an interface that needs to be registered as its first argument and a function/method as that will resolve and return an instance of this class.

from expanse.container.container import Container


def create_message_generator(container: Container) -> MessageGenerator:
    return MessageGenerator(container.get(MessagePrinter))


container = Container()

container.register(
    MessageGenerator, create_message_generator
)

The resolving function can be any callable that returns an instance of the class that is being bound. Any dependencies declared in the resolver function will be resolved automatically by the container, so the example above could be rewritten as:

def create_message_generator(printer: MessagePrinter) -> MessageGenerator:
    return MessageGenerator(printer)

For simple cases, you can directly register the class itself instead: the needed dependencies will be resolved automatically.

container.register(MessageGenerator)

Singletons

If a service should only be resolved once during the lifecycle of the application, you can use the singleton() method. A singleton service will be resolved the first time it is requested and the same instance will be returned on subsequent calls.

container.singleton(
    MessageGeneratorInterface, create_message_generator
)

Scoped singletons

If a service should only be resolved once during the lifecycle of a request, you can use the scoped() method. It's similar to the singleton() method but the resolved instances will be discarded at the end of the request.

container.scoped(
    MessageGeneratorInterface, create_message_generator
)

Instances

If you already have an instance of a service you want to register in the container, you can use the instance() method.

generator = MessageGenerator(MessagePrinter())

container.instance(MessageGeneratorInterface, generator)

Resolving services

Resolving services manually

If you want to resolve a service directly from the container, you can use the make() method. It accepts either the name of the class or interface you want to resolve.

generator = container.get(MessageGeneratorInterface)

If you need to know if a service is bound to the container before resolving it, you can use the has() method.

if container.has(MessageGenerator):
    generator = container.get(MessageGenerator)

Automatic service injection

Most of the time you will not have to interact with the container directly to resolver the services you need. Whether it's in your controllers, middleware, or commands, you can type-hint the dependencies you need and they will be resolved automatically for you.

Let's say you manage your users with a UsersRepository class, you can inject an instance of it inside the controller by simply type-hinting it:

from expanse.view.view import View
from expanse.view.view_factory import ViewFactory

from app.repositories.users_repository import UsersRepository


class UsersController:

    def __init__(self, repository: UsersRepository) -> None:
        self._repository = repository

    def index(self, view: ViewFactory) -> View:
        users = self._repository.get_all_users()

        return view.make(
            "users/index", {"users": users}
        )

Calling methods or functions and injecting services

Sometimes you may need to call a method or a function not configured for automatic injection, but still leverage the container to resolve the needed services. Let's say you have a class dedicated to generating reports about your users:

class UserReport:

    def generate(self, repository: UsersRepository) -> list:
        users = repository.get_all_users()

        return [
            ...,
        ]

You can call the generate() method by using the call() method of the container.

container.call(UserReport().generate)

Terminating services

Some services will need to be cleaned up when the container is terminated, either at the end of a request for scoped services or at the end of the application lifecycle. This is notably the case of database connections where you need to make sure that they are properly closed.

There are two ways to make sure a service is cleanly terminated: either by registering explicitly a termination callback that will be called when the container is terminated, or by registering the service via a generator function/method.

Termination callbacks

You can register a termination callback for a service by using the terminating() method.

def close_connection(connection: Connection) -> None:
    connection.close()


container.terminating(Connection, close_connection)

This is all you need. When the container is terminated, close_connection() will be automatically closed.

Using generator functions

Registering explicit termination callbacks can be cumbersome, especially if you have many services that need to be cleaned up. A simpler alternative is to use generator functions or methods instead.

from collections.abc import Generator


def create_connection(db: DatabaseManager) -> Generator[Connection, None, None]:
    connection = db.connection()

    yield connection

    connection.close()


container.scoped(Connection, create_connection)

When the connection is resolved for the first time, the generator will be called and will yield the connection, and a function will be automatically be registered as a terminating callback that will close the connection.