Application example (multiple containers) — Dependency Injector 4.46.0 documentation (original) (raw)

This example shows how you can create an application using multiple declarative containers. Using multiple declarative containers is a good choice for a large application. For building a moderate or a small size application refer to Application example (single container).

We build an example micro application following the dependency injection principle. It consists of several services with a domain logic. The services have dependencies on database & AWS S3.

Start from the scratch or jump to the section:

• Application structure
• Containers
• Main module
• Services
• Configuration
• Run the application

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

./ ├── example/ │ ├── init.py │ ├── main.py │ ├── containers.py │ └── services.py ├── config.yml └── requirements.txt

Containers¶

Listing of the example/containers.py:

"""Containers module."""

import logging.config import sqlite3

import boto3 from dependency_injector import containers, providers

from . import services

class Core(containers.DeclarativeContainer):

config = providers.Configuration()

logging = providers.Resource(
    logging.config.dictConfig,
    config=config.logging,
)

class Gateways(containers.DeclarativeContainer):

config = providers.Configuration()

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

s3_client = 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,
)

class Services(containers.DeclarativeContainer):

config = providers.Configuration()
gateways = providers.DependenciesContainer()

user = providers.Factory(
    services.UserService,
    db=gateways.database_client,
)

auth = providers.Factory(
    services.AuthService,
    db=gateways.database_client,
    token_ttl=config.auth.token_ttl.as_int(),
)

photo = providers.Factory(
    services.PhotoService,
    db=gateways.database_client,
    s3=gateways.s3_client,
)

class Application(containers.DeclarativeContainer):

config = providers.Configuration(yaml_files=["config.yml"])

core = providers.Container(
    Core,
    config=config.core,
)

gateways = providers.Container(
    Gateways,
    config=config.gateways,
)

services = providers.Container(
    Services,
    config=config.services,
    gateways=gateways,
)

Main module¶

Listing of the example/__main__.py:

"""Main module."""

import sys

from dependency_injector.wiring import Provide, inject

from .services import UserService, AuthService, PhotoService from .containers import Application

@inject def main( email: str, password: str, photo: str, user_service: UserService = Provide[Application.services.user], auth_service: AuthService = Provide[Application.services.auth], photo_service: PhotoService = Provide[Application.services.photo], ) -> None: user = user_service.get_user(email) auth_service.authenticate(user, password) photo_service.upload_photo(user, photo)

if name == "main": application = Application() application.core.init_resources() application.wire(modules=[name])

main(*sys.argv[1:])

Services¶

Listing of the example/services.py:

"""Services module."""

import logging import sqlite3 from typing import Dict

from mypy_boto3_s3 import S3Client

class BaseService:

def __init__(self) -> None:
    self.logger = logging.getLogger(
        f"{__name__}.{self.__class__.__name__}",
    )

class UserService(BaseService):

def __init__(self, db: sqlite3.Connection) -> None:
    self.db = db
    super().__init__()

def get_user(self, email: str) -> Dict[str, str]:
    self.logger.debug("User %s has been found in database", email)
    return {"email": email, "password_hash": "..."}

class AuthService(BaseService):

def __init__(self, db: sqlite3.Connection, token_ttl: int) -> None:
    self.db = db
    self.token_ttl = token_ttl
    super().__init__()

def authenticate(self, user: Dict[str, str], password: str) -> None:
    assert password is not None
    self.logger.debug(
        "User %s has been successfully authenticated",
        user["email"],
    )

class PhotoService(BaseService):

def __init__(self, db: sqlite3.Connection, s3: S3Client) -> None:
    self.db = db
    self.s3 = s3
    super().__init__()

def upload_photo(self, user: Dict[str, str], photo_path: str) -> None:
    self.logger.debug(
        "Photo %s has been successfully uploaded by user %s",
        photo_path,
        user["email"],
    )

Configuration¶

Listing of the config.yml:

core:

logging: version: 1 formatters: formatter: format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s" handlers: console: class: "logging.StreamHandler" level: "DEBUG" formatter: "formatter" stream: "ext://sys.stderr" root: level: "DEBUG" handlers: ["console"]

gateways:

database: dsn: ":memory:"

aws: access_key_id: "KEY" secret_access_key: "SECRET"

services:

auth: token_ttl: 3600

Run the application¶

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