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.

../_images/application.png

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."""
importlogging.config
importsqlite3
importboto3
fromdependency_injectorimport containers, providers
from.import services
classCore(containers.DeclarativeContainer):
 config = providers.Configuration()
 logging = providers.Resource(
 logging.config.dictConfig,
 config=config.logging,
 )
classGateways(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,
 )
classServices(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,
 )
classApplication(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."""
importsys
fromdependency_injector.wiringimport Provide, inject
from.servicesimport UserService, AuthService, PhotoService
from.containersimport Application
@inject
defmain(
 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."""
importlogging
importsqlite3
fromtypingimport Dict
frommypy_boto3_s3import S3Client
classBaseService:
 def__init__(self) -> None:
 self.logger = logging.getLogger(
 f"{__name__}.{self.__class__.__name__}",
 )
classUserService(BaseService):
 def__init__(self, db: sqlite3.Connection) -> None:
 self.db = db
 super().__init__()
 defget_user(self, email: str) -> Dict[str, str]:
 self.logger.debug("User %s has been found in database", email)
 return {"email": email, "password_hash": "..."}
classAuthService(BaseService):
 def__init__(self, db: sqlite3.Connection, token_ttl: int) -> None:
 self.db = db
 self.token_ttl = token_ttl
 super().__init__()
 defauthenticate(self, user: Dict[str, str], password: str) -> None:
 assert password is not None
 self.logger.debug(
 "User %s has been successfully authenticated",
 user["email"],
 )
classPhotoService(BaseService):
 def__init__(self, db: sqlite3.Connection, s3: S3Client) -> None:
 self.db = db
 self.s3 = s3
 super().__init__()
 defupload_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:

[フレーム]