Refactor services mini app with single container

examples/miniapps/services-single-container/README.rst

@@ -0,0 +1,29 @@
+Services mini application example (single container)
+====================================================
+
+Create virtual env:
+
+.. code-block:: bash
+
+   python3 -m venv venv
+   . venv/bin/activate
+
+Install requirements:
+
+.. code-block:: bash
+
+   pip install -r requirements.txt
+
+Run:
+
+.. code-block:: bash
+
+   python -m example user@example.com secret photo.jpg
+
+You should see:
+
+.. code-block:: bash
+
+   [2020-09-04 15:27:27,727] [DEBUG] [example.services.UserService]: User user@example.com has been found in database
+   [2020-09-04 15:27:27,727] [DEBUG] [example.services.AuthService]: User user@example.com has been successfully authenticated
+   [2020-09-04 15:27:27,727] [DEBUG] [example.services.PhotoService]: Photo photo.jpg has been successfully uploaded by user user@example.com

examples/miniapps/services-single-container/config.ini

@@ -0,0 +1,9 @@
+[database]
+dsn=:memory:
+
+[aws]
+access_key_id=KEY
+secret_access_key=SECRET
+
+[auth]
+token_ttl=3600

examples/miniapps/services-single-container/example/__main__.py

@@ -0,0 +1,24 @@
+"""Main module."""
+
+import sys
+
+from .containers import Container
+
+
+def main(email: str, password: str, photo: str) -> None:
+    container = Container()
+
+    container.configure_logging()
+    container.config.from_ini('config.ini')
+
+    user_service = container.user_service()
+    auth_service = container.auth_service()
+    photo_service = container.photo_service()
+
+    user = user_service.get_user(email)
+    auth_service.authenticate(user, password)
+    photo_service.upload_photo(user, photo)
+
+
+if __name__ == '__main__':
+    main(*sys.argv[1:])

examples/miniapps/services-single-container/example/containers.py

@@ -0,0 +1,52 @@
+"""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('config')
+
+    configure_logging = providers.Callable(
+        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,
+    )

examples/miniapps/services-single-container/example/services.py

@@ -0,0 +1,56 @@
+"""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'],
+        )

examples/miniapps/services-single-container/logging.ini

@@ -0,0 +1,21 @@
+[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

examples/miniapps/services-single-container/requirements.txt

@@ -0,0 +1,2 @@
+boto3
+boto3-stubs[s3]

examples/miniapps/services_v2/README.rst

@@ -1,8 +0,0 @@
-Dependency Injector IoC containers example
-==========================================
-
-Instructions for running
-
-.. code-block:: bash
-
-    python run.py 1 secret photo.jpg

examples/miniapps/services_v2/container.py

@@ -1,57 +0,0 @@
-"""Example of dependency injection in Python."""
-
-import logging
-import sqlite3
-
-import boto3
-
-from dependency_injector import containers, providers
-from example import services, main
-
-
-class IocContainer(containers.DeclarativeContainer):
-    """Application IoC container."""
-
-    config = providers.Configuration('config')
-    logger = providers.Singleton(logging.Logger, name='example')
-
-    # Gateways
-
-    database_client = providers.Singleton(sqlite3.connect, config.database.dsn)
-
-    s3_client = providers.Singleton(
-        boto3.client, 's3',
-        aws_access_key_id=config.aws.access_key_id,
-        aws_secret_access_key=config.aws.secret_access_key,
-    )
-
-    # Services
-
-    users_service = providers.Factory(
-        services.UsersService,
-        db=database_client,
-        logger=logger,
-    )
-
-    auth_service = providers.Factory(
-        services.AuthService,
-        token_ttl=config.auth.token_ttl,
-        db=database_client,
-        logger=logger,
-    )
-
-    photos_service = providers.Factory(
-        services.PhotosService,
-        db=database_client,
-        s3=s3_client,
-        logger=logger,
-    )
-
-    # Misc
-
-    main = providers.Callable(
-        main.main,
-        users_service=users_service,
-        auth_service=auth_service,
-        photos_service=photos_service,
-    )

examples/miniapps/services_v2/example/main.py

@@ -1,8 +0,0 @@
-"""Example main module."""
-
-
-def main(uid, password, photo, users_service, auth_service, photos_service):
-    """Authenticate user and upload photo."""
-    user = users_service.get_user_by_id(uid)
-    auth_service.authenticate(user, password)
-    photos_service.upload_photo(user['uid'], photo)

examples/miniapps/services_v2/example/services.py

@@ -1,50 +0,0 @@
-"""Example business services module."""
-
-
-class BaseService:
-    """Service base class."""
-
-
-class UsersService(BaseService):
-    """Users service."""
-
-    def __init__(self, logger, db):
-        """Initialize instance."""
-        self.logger = logger
-        self.db = db
-
-    def get_user_by_id(self, uid):
-        """Return user's data by identifier."""
-        self.logger.debug('User %s has been found in database', uid)
-        return dict(uid=uid, password_hash='secret_hash')
-
-
-class AuthService(BaseService):
-    """Authentication service."""
-
-    def __init__(self, logger, db, token_ttl):
-        """Initialize instance."""
-        self.logger = logger
-        self.db = db
-        self.token_ttl = token_ttl
-
-    def authenticate(self, user, password):
-        """Authenticate user."""
-        assert user['password_hash'] == '_'.join((password, 'hash'))
-        self.logger.debug('User %s has been successfully authenticated',
-                          user['uid'])
-
-
-class PhotosService(BaseService):
-    """Photos service."""
-
-    def __init__(self, logger, db, s3):
-        """Initialize instance."""
-        self.logger = logger
-        self.db = db
-        self.s3 = s3
-
-    def upload_photo(self, uid, photo_path):
-        """Upload user photo."""
-        self.logger.debug('Photo %s has been successfully uploaded by user %s',
-                          photo_path, uid)

examples/miniapps/services_v2/run.py

@@ -1,28 +0,0 @@
-"""Run example of dependency injection in Python."""
-
-import sys
-import logging
-
-from container import IocContainer
-
-
-if __name__ == '__main__':
-    # Configure container:
-    container = IocContainer(
-        config={
-            'database': {
-                'dsn': ':memory:',
-            },
-            'aws': {
-                'access_key_id': 'KEY',
-                'secret_access_key': 'SECRET',
-            },
-            'auth': {
-                'token_ttl': 3600,
-            },
-        }
-    )
-    container.logger().addHandler(logging.StreamHandler(sys.stdout))
-
-    # Run application:
-    container.main(*sys.argv[1:])