Services mini application example (v1 - multiple containers)

“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
"""Example business services module."""


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


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

    def __init__(self, logger, db):
        """Initializer."""
        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):
        """Initializer."""
        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):
        """Initializer."""
        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)

Listing of example/main.py:

1
2
3
4
5
6
7
8
"""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)

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.override({'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])