python-dependency-injector/examples/providers/external_dependency.py

"""`ExternalDependency` providers example."""

import sqlite3
import contextlib

import dependency_injector.providers as providers


class UsersService(object):
    """Example class UsersService.

    UsersService has dependency on DBAPI 2.0 database connection.
    """

    def __init__(self, database):
        """Initializer.

        :param database: Database connection.
        :type database: sqlite3.dbapi2.Connection
        """
        self.database = database
        self.database.row_factory = sqlite3.dbapi2.Row

    def init_database(self):
        """Initialize database, if it has not been initialized yet."""
        with contextlib.closing(self.database.cursor()) as cursor:
            cursor.execute("""
              CREATE TABLE IF NOT EXISTS users(
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name VARCHAR(32)
              )
            """)

    def create(self, name):
        """Create user with provided name and return his id."""
        with contextlib.closing(self.database.cursor()) as cursor:
            cursor.execute('INSERT INTO users(name) VALUES (?)', (name,))
            return cursor.lastrowid

    def get_by_id(self, id):
        """Return user info by user id."""
        with contextlib.closing(self.database.cursor()) as cursor:
            cursor.execute('SELECT id, name FROM users WHERE id=?', (id,))
            return cursor.fetchone()


# Database and UsersService providers:
database = providers.ExternalDependency(instance_of=sqlite3.dbapi2.Connection)
users_service_factory = providers.Factory(UsersService,
                                          database=database)

# Out of library's scope.
#
# Setting database provider:
database.provided_by(providers.Singleton(sqlite3.dbapi2.Connection,
                                         database=':memory:',
                                         timeout=30,
                                         detect_types=True,
                                         isolation_level='EXCLUSIVE'))

# Creating UsersService instance:
users_service = users_service_factory()

# Initializing UsersService database:
users_service.init_database()

# Creating test user and retrieving full information about him:
test_user_id = users_service.create(name='test_user')
test_user = users_service.get_by_id(test_user_id)

# Making some asserts:
assert test_user['id'] == 1
assert test_user['name'] == 'test_user'