Application example (multiple containers)

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:

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/ │ ├── __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.

Sponsor the project on GitHub: