Application example (single container) — Dependency Injector 4.46.0 documentation (original) (raw)
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.
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 arequirements.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."""
import logging.config import sqlite3
import boto3 from dependency_injector import containers, providers
from . import services
class Container(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."""
import sys
from dependency_injector.wiring import Provide, inject
from .services import UserService, AuthService, PhotoService from .containers import Container
@inject def main( 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."""
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.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: |
---|