Decoupled packages example (multiple containers)

This example shows how to use Dependency Injector to create decoupled packages.

To achieve a decoupling each package has a container with the components. When a component needs a dependency from the outside of the package scope we use the Dependency provider. The package container has no knowledge on where the dependencies come from. It states a need that the dependencies must be provided. This helps to decouple a package from the 3rd party dependencies and other packages.

To wire the packages we use an application container. Application container has all 3rd party dependencies and package containers. It wires the packages and dependencies to create a complete application.

We build an example micro application that consists of 3 packages:

• user - a package with user domain logic, depends on a database

• photo - a package with photo domain logic, depends on a database and AWS S3

• analytics - a package with analytics domain logic, depends on the user and photo package components

Start from the scratch or jump to the section:

You can find the source code and instructions for running on the Github.

Application structure

Application consists of an example package, a configuration file and a requirements.txt file.

./
├── example/
│   ├── analytics/
│   │   ├── __init__.py
│   │   ├── containers.py
│   │   └── services.py
│   ├── photo/
│   │   ├── __init__.py
│   │   ├── containers.py
│   │   ├── entities.py
│   │   └── repositories.py
│   ├── user/
│   │   ├── __init__.py
│   │   ├── containers.py
│   │   ├── entities.py
│   │   └── repositories.py
│   ├── __init__.py
│   ├── __main__.py
│   └── containers.py
├── config.ini
└── requirements.txt

Package containers

Listing of the example/user/containers.py:

"""User containers module."""

from dependency_injector import containers, providers

from . import entities, repositories


class UserContainer(containers.DeclarativeContainer):

    database = providers.Dependency()

    user = providers.Factory(entities.User)

    user_repository = providers.Singleton(
        repositories.UserRepository,
        entity_factory=user.provider,
        db=database,
    )

Listing of the example/photo/containers.py:

"""Photo containers module."""

from dependency_injector import containers, providers

from . import entities, repositories


class PhotoContainer(containers.DeclarativeContainer):

    database = providers.Dependency()
    file_storage = providers.Dependency()

    photo = providers.Factory(entities.Photo)

    photo_repository = providers.Singleton(
        repositories.PhotoRepository,
        entity_factory=photo.provider,
        fs=file_storage,
        db=database,
    )

Listing of the example/analytics/containers.py:

"""Analytics containers module."""

from dependency_injector import containers, providers

from . import services


class AnalyticsContainer(containers.DeclarativeContainer):

    user_repository = providers.Dependency()
    photo_repository = providers.Dependency()

    aggregation_service = providers.Singleton(
        services.AggregationService,
        user_repository=user_repository,
        photo_repository=photo_repository,
    )

Application container

Application container consists of all packages and 3rd party dependencies. Its role is to wire everything together in a complete application.

Listing of the example/containers.py:

"""Containers module."""

import sqlite3

import boto3
from dependency_injector import containers, providers

from .user.containers import UserContainer
from .photo.containers import PhotoContainer
from .analytics.containers import AnalyticsContainer


class ApplicationContainer(containers.DeclarativeContainer):

    config = providers.Configuration(ini_files=["config.ini"])

    sqlite = providers.Singleton(sqlite3.connect, config.database.dsn)

    s3 = providers.Singleton(
        boto3.client,
        service_name="s3",
        aws_access_key_id=config.aws.access_key_id,
        aws_secret_access_key=config.aws.secret_access_key,
    )

    user_package = providers.Container(
        UserContainer,
        database=sqlite,
    )

    photo_package = providers.Container(
        PhotoContainer,
        database=sqlite,
        file_storage=s3,
    )

    analytics_package = providers.Container(
        AnalyticsContainer,
        user_repository=user_package.user_repository,
        photo_repository=photo_package.photo_repository,
    )

Note

Package analytics has dependencies on the repositories from the user and photo packages. This is an example of how you can pass the dependencies from one package to another.

Main module

Listing of the example/__main__.py:

"""Main module."""

from dependency_injector.wiring import Provide, inject

from .user.repositories import UserRepository
from .photo.repositories import PhotoRepository
from .analytics.services import AggregationService
from .containers import ApplicationContainer


@inject
def main(
        user_repository: UserRepository = Provide[
            ApplicationContainer.user_package.user_repository
        ],
        photo_repository: PhotoRepository = Provide[
            ApplicationContainer.photo_package.photo_repository
        ],
        aggregation_service: AggregationService = Provide[
            ApplicationContainer.analytics_package.aggregation_service
        ],
) -> None:
    user1 = user_repository.get(id=1)
    user1_photos = photo_repository.get_photos(user1.id)
    print(f"Retrieve user id={user1.id}, photos count={len(user1_photos)}")

    user2 = user_repository.get(id=2)
    user2_photos = photo_repository.get_photos(user2.id)
    print(f"Retrieve user id={user2.id}, photos count={len(user2_photos)}")

    assert aggregation_service.user_repository is user_repository
    assert aggregation_service.photo_repository is photo_repository
    print("Aggregate analytics from user and photo packages")


if __name__ == "__main__":
    application = ApplicationContainer()
    application.wire(modules=[__name__])

    main()

Configuration

Listing of the config.ini:

[database]
dsn=:memory:

[aws]
access_key_id=KEY
secret_access_key=SECRET

Run the application

You can find the source code and instructions for running on the Github.

Sponsor the project on GitHub: