Application example (single container)¶

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

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, several configuration files and a requirements.txt file.

./
├──example/
│├──__init__.py
│├──__main__.py
│├──containers.py
│└──services.py
├──config.ini
├──logging.ini
└──requirements.txt

Container¶

Listing of the example/containers.py:

"""Containers module."""
importlogging.config
importsqlite3
importboto3
fromdependency_injectorimport containers, providers
from.import services
classContainer(containers.DeclarativeContainer):
 config = providers.Configuration(ini_files=["config.ini"])
 logging = providers.Resource(
 logging.config.fileConfig,
 fname="logging.ini",
 )
 # Gateways
 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,
 )
 # Services
 user_service = providers.Factory(
 services.UserService,
 db=database_client,
 )
 auth_service = providers.Factory(
 services.AuthService,
 db=database_client,
 token_ttl=config.auth.token_ttl.as_int(),
 )
 photo_service = providers.Factory(
 services.PhotoService,
 db=database_client,
 s3=s3_client,
 )

Main module¶

Listing of the example/__main__.py:

"""Main module."""
importsys
fromdependency_injector.wiringimport Provide, inject
from.servicesimport UserService, AuthService, PhotoService
from.containersimport Container
@inject
defmain(
 email: str,
 password: str,
 photo: str,
 user_service: UserService = Provide[Container.user_service],
 auth_service: AuthService = Provide[Container.auth_service],
 photo_service: PhotoService = Provide[Container.photo_service],
) -> None:
 user = user_service.get_user(email)
 auth_service.authenticate(user, password)
 photo_service.upload_photo(user, photo)
if __name__ == "__main__":
 container = Container()
 container.init_resources()
 container.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.ini:

[database]
dsn=:memory:
[aws]
access_key_id=KEY
secret_access_key=SECRET
[auth]
token_ttl=3600

Listing of the logging.ini:

[loggers]
keys=root
[handlers]
keys=stream_handler
[formatters]
keys=formatter
[logger_root]
level=DEBUG
handlers=stream_handler
[handler_stream_handler]
class=StreamHandler
level=DEBUG
formatter=formatter
args=(sys.stderr,)
[formatter_formatter]
format=[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s

Run the application¶

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

Sponsor the project on GitHub:

[フレーム]