Services mini application example

“Services miniapp” is an example mini application that consists from several services that have dependencies on some standard and 3rd-party libraries for logging, interaction with database and remote service calls via API.

“Services miniapp” example demonstrates usage of Dependency Injector for creating several IoC containers.

Instructions for running:

python run.py 1 secret photo.jpg

Example application

Classes diagram:

../_images/classes1.png

Example application structure:

/example
    /__init__.py
    /main.py
    /services.py

Listing of example/services.py:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
"""Example business services module."""


class BaseService(object):
    """Service base class."""


class UsersService(BaseService):
    """Users service."""

    def __init__(self, logger, db):
        """Initializer.

        :param logger: Logger instance.
        :type logger: logging.Logger

        :param db: Database connection.
        :type db: sqlite3.Connection
        """
        self.logger = logger
        self.db = db

    def get_user_by_id(self, uid):
        """Return user's data by identifier.

        :param uid: User identifier.
        :type uid: int

        :rtype: dict
        """
        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):
        """Initializer.

        :param logger: Logger instance.
        :type logger: logging.Logger

        :param db: Database connection.
        :type db: sqlite3.Connection

        :param token_ttl: Token lifetime in seconds.
        :type token_ttl: int
        """
        self.logger = logger
        self.db = db
        self.token_ttl = token_ttl

    def authenticate(self, user, password):
        """Authenticate user.

        :param user: User's data.
        :type user: dict

        :param password: User's password for verification.
        :type password: str

        :raises: AssertionError when password is wrong

        :rtype: None
        """
        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):
        """Initializer.

        :param logger: Logger instance.
        :type logger: logging.Logger

        :param db: Database connection.
        :type db: sqlite3.Connection

        :param s3: AWS S3 client.
        :type s3: botocore.client.S3
        """
        self.logger = logger
        self.db = db
        self.s3 = s3

    def upload_photo(self, uid, photo_path):
        """Upload user photo.

        :param uid: User identifier.
        :type uid: int

        :param photo_path: Path to photo for uploading.
        :type photo_path: str

        :rtpe: None
        """
        self.logger.debug('Photo %s has been successfully uploaded by user %s',
                          photo_path, uid)

Listing of example/main.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
"""Example main module."""


def main(uid, password, photo, users_service, auth_service, photos_service):
    """Authenticate user and upload photo.

    :param uid: User identifier.
    :type uid: int

    :param password: User's password for verification.
    :type password: str

    :param photo_path: Path to photo for uploading.
    :type photo_path: str

    :param users_service: Users service.
    :type users_service: example.services.UsersService

    :param auth_service: Authentication service.
    :type auth_service: example.services.AuthService

    :param photo_service: Photo service.
    :type photo_service: example.services.PhotoService
    """
    user = users_service.get_user_by_id(uid)
    auth_service.authenticate(user, password)
    photos_service.upload_photo(user['uid'], photo)

IoC containers

Listing of containers.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
"""Example of dependency injection in Python."""

import logging
import sqlite3

import boto3

import example.main
import example.services

import dependency_injector.containers as containers
import dependency_injector.providers as providers


class Core(containers.DeclarativeContainer):
    """IoC container of core component providers."""

    config = providers.Configuration('config')

    logger = providers.Singleton(logging.Logger, name='example')


class Gateways(containers.DeclarativeContainer):
    """IoC container of gateway (API clients to remote services) providers."""

    database = providers.Singleton(sqlite3.connect, Core.config.database.dsn)

    s3 = providers.Singleton(
        boto3.client, 's3',
        aws_access_key_id=Core.config.aws.access_key_id,
        aws_secret_access_key=Core.config.aws.secret_access_key)


class Services(containers.DeclarativeContainer):
    """IoC container of business service providers."""

    users = providers.Factory(example.services.UsersService,
                              db=Gateways.database,
                              logger=Core.logger)

    auth = providers.Factory(example.services.AuthService,
                             db=Gateways.database,
                             logger=Core.logger,
                             token_ttl=Core.config.auth.token_ttl)

    photos = providers.Factory(example.services.PhotosService,
                               db=Gateways.database,
                               s3=Gateways.s3,
                               logger=Core.logger)


class Application(containers.DeclarativeContainer):
    """IoC container of application component providers."""

    main = providers.Callable(example.main.main,
                              users_service=Services.users,
                              auth_service=Services.auth,
                              photos_service=Services.photos)

Run application

Listing of run.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
"""Run example application."""

import sys
import logging

from containers import Core, Application


if __name__ == '__main__':
    # Configure platform:
    Core.config.update({'database': {'dsn': ':memory:'},
                        'aws': {'access_key_id': 'KEY',
                                'secret_access_key': 'SECRET'},
                        'auth': {'token_ttl': 3600}})
    Core.logger().addHandler(logging.StreamHandler(sys.stdout))

    # Run application:
    Application.main(uid=sys.argv[1],
                     password=sys.argv[2],
                     photo=sys.argv[3])