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

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

from dependency_injector.providers import ExternalDependency
from dependency_injector.providers import Factory
from dependency_injector.providers import Singleton

from dependency_injector.injections import KwArg
from dependency_injector.injections import Attribute

# Importing SQLITE3 and contextlib.closing for working with cursors:
import sqlite3
from contextlib import closing


# Definition of example UserService:
class UserService(object):

    """Example class UserService.

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

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

        Database dependency need to be injected via init arg.
        """
        self.database = database

    def init_database(self):
        """Initialize database, if it has not been initialized yet."""
        with 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 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 closing(self.database.cursor()) as cursor:
            cursor.execute('SELECT id, name FROM users WHERE id=?', (id,))
            return cursor.fetchone()


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

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

# Creating UserService instance:
users_service = users_service_factory()

# Initializing UserService 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'