Implementing Dependency Injection in Python Flask Using Dependency Injector

github project link : https://github.com/dkmostafa/python-flask-dependency-injector-sample

Objective :

The objective is to apply the dependency injection pattern to our Python Flask application using the dependency_injector package.

Installation :

Begin by installing the dependency-injector package, which is listed in the requirements.txt file

Creating our Application Container :

To begin, we create an application container to manage our required services, data repositories, and database configuration. Below is a sample code snippet:

employee_container.py

from dependency_injector import containers, providers
from ..db.database import Database
from ..db.repositories.employee_repository import EmployeeRepository
from .controllers.employee_controller import EmployeeController
from .services.employee_service import EmployeeService
class EmployeeRepositories(containers.DeclarativeContainer):
    db_url = "sqlite:///employees.db"
    db = providers.Singleton(Database, db_url=db_url)
    db_session = db.provided.session
    employee_repository = providers.Factory(EmployeeRepository, session_factory=db_session)
class EmployeeServices(containers.DeclarativeContainer):
    employee_service = providers.Factory(EmployeeService, employee_repository=EmployeeRepositories.employee_repository)
class EmployeeControllers(containers.DeclarativeContainer):
    employee_controller = providers.Factory(EmployeeController, employee_service=EmployeeServices.employee_service)
class EmployeeContainer(containers.DeclarativeContainer):
    repositories = providers.Container(EmployeeRepositories)
    services = providers.Container(EmployeeServices)
    controllers = providers.Container(EmployeeControllers)
from dependency_injector import containers, providers
from ..db.database import Database
from ..db.repositories.employee_repository import EmployeeRepository
from .controllers.employee_controller import EmployeeController
from .services.employee_service import EmployeeService

class EmployeeRepositories(containers.DeclarativeContainer):
    db_url = "sqlite:///employees.db"
    db = providers.Singleton(Database, db_url=db_url)
    db_session = db.provided.session
    employee_repository = providers.Factory(EmployeeRepository, session_factory=db_session)

class EmployeeServices(containers.DeclarativeContainer):
    employee_service = providers.Factory(EmployeeService, employee_repository=EmployeeRepositories.employee_repository)

class EmployeeControllers(containers.DeclarativeContainer):
    employee_controller = providers.Factory(EmployeeController, employee_service=EmployeeServices.employee_service)

class EmployeeContainer(containers.DeclarativeContainer):
    repositories = providers.Container(EmployeeRepositories)
    services = providers.Container(EmployeeServices)
    controllers = providers.Container(EmployeeControllers)
from dependency_injector import containers, providers
from ..db.database import Database
from ..db.repositories.employee_repository import EmployeeRepository
from .controllers.employee_controller import EmployeeController
from .services.employee_service import EmployeeService

class EmployeeRepositories(containers.DeclarativeContainer):
    db_url = "sqlite:///employees.db"
    db = providers.Singleton(Database, db_url=db_url)
    db_session = db.provided.session
    employee_repository = providers.Factory(EmployeeRepository, session_factory=db_session)

class EmployeeServices(containers.DeclarativeContainer):
    employee_service = providers.Factory(EmployeeService, employee_repository=EmployeeRepositories.employee_repository)

class EmployeeControllers(containers.DeclarativeContainer):
    employee_controller = providers.Factory(EmployeeController, employee_service=EmployeeServices.employee_service)

class EmployeeContainer(containers.DeclarativeContainer):
    repositories = providers.Container(EmployeeRepositories)
    services = providers.Container(EmployeeServices)
    controllers = providers.Container(EmployeeControllers)

Enter fullscreen mode Exit fullscreen mode

This code organizes our application’s components into containers (EmployeeRepositories, EmployeeServices, EmployeeControllers, and EmployeeContainer). It sets up dependencies such as database connections (Database) and services (EmployeeService) using the dependency_injector framework in Python Flask.

Wiring the created container to our application with the routes file:

src/app.py

from flask import Flask
from src.modules.employee import employee_routes
from src.modules.employee.employee_container import EmployeeContainer
app = Flask(__name__)
app.register_blueprint(employee_routes.employee_blueprint)
employee_container = EmployeeContainer()
employee_container.wire(
    modules=[employee_routes.__name__]
)
if __name__ == "__main__":
    app.run(debug=True)
from flask import Flask
from src.modules.employee import employee_routes
from src.modules.employee.employee_container import EmployeeContainer

app = Flask(__name__)

app.register_blueprint(employee_routes.employee_blueprint)

employee_container = EmployeeContainer()

employee_container.wire(
    modules=[employee_routes.__name__]
)

if __name__ == "__main__":
    app.run(debug=True)
from flask import Flask
from src.modules.employee import employee_routes
from src.modules.employee.employee_container import EmployeeContainer

app = Flask(__name__)

app.register_blueprint(employee_routes.employee_blueprint)

employee_container = EmployeeContainer()

employee_container.wire(
    modules=[employee_routes.__name__]
)

if __name__ == "__main__":
    app.run(debug=True)

Enter fullscreen mode Exit fullscreen mode

After wiring our container to our app and routes created , we can start injecting the created services , repositories in our application as following :

Injecting controller and using it :

employee_controller: EmployeeController = Provide[EmployeeContainer.controllers.employee_controller]
employee_controller.create_employee(body)
employee_controller: EmployeeController = Provide[EmployeeContainer.controllers.employee_controller]

employee_controller.create_employee(body)
employee_controller: EmployeeController = Provide[EmployeeContainer.controllers.employee_controller]

employee_controller.create_employee(body)

Enter fullscreen mode Exit fullscreen mode

Injecting EmployeeService into EmployeeController :

class EmployeeController:
    def __init__(self, employee_service: EmployeeService):
        self.employee_service = employee_service
        pass;
class EmployeeController:

    def __init__(self, employee_service: EmployeeService):
        self.employee_service = employee_service
        pass;
class EmployeeController:

    def __init__(self, employee_service: EmployeeService):
        self.employee_service = employee_service
        pass;

Enter fullscreen mode Exit fullscreen mode

Testing when using dependency injection :

Using dependency injection simplifies and enhances the manageability of testing and mocking each service in our application.

With dependency injection, our services and controllers receive their dependencies through constructors or method parameters, making it straightforward to replace real dependencies with mocks or stubs during testing. This approach improves test isolation and ensures that tests focus solely on the behavior of the unit under test.

Here’s an example of how testing might look with dependency injection in a Python Flask application:

import pytest
from unittest.mock import MagicMock
from ..employee_service import EmployeeService
from ...controllers.dtos.employee_controller_dto import CreateEmployeeDTO
@pytest.fixture
def employee_service():
    mock_employee_repository = MagicMock()
    mock_employee_repository.save.return_value = {
        "id": 1,
        "name": "test",
        "username": "test",
        "email": "test",
    }
    mock_employee_repository.get_all_employees.return_value = [
        {
            "email": "test",
            "id": 1,
            "name": "test",
            "username": "test"
        }
    ]
    employee_service = EmployeeService(mock_employee_repository)
    return employee_service
def test_create_employee(employee_service):
    create_employee_mock_input = CreateEmployeeDTO(
        name="test",
        username="test",
        email="test",
    )
    res = employee_service.create_employee(create_employee_mock_input)
    assert res == {
        "id": 1,
        "name": "test",
        "username": "test",
        "email": "test",
    }
    assert True
def test_get_employees(employee_service):
    res = employee_service.get_all_employees()
    assert res == [
        {
            "email": "test",
            "id": 1,
            "name": "test",
            "username": "test"
        }
    ]
import pytest
from unittest.mock import MagicMock
from ..employee_service import EmployeeService
from ...controllers.dtos.employee_controller_dto import CreateEmployeeDTO


@pytest.fixture
def employee_service():
    mock_employee_repository = MagicMock()

    mock_employee_repository.save.return_value = {
        "id": 1,
        "name": "test",
        "username": "test",
        "email": "test",
    }
    mock_employee_repository.get_all_employees.return_value = [
        {
            "email": "test",
            "id": 1,
            "name": "test",
            "username": "test"
        }
    ]

    employee_service = EmployeeService(mock_employee_repository)

    return employee_service


def test_create_employee(employee_service):
    create_employee_mock_input = CreateEmployeeDTO(
        name="test",
        username="test",
        email="test",
    )
    res = employee_service.create_employee(create_employee_mock_input)

    assert res == {
        "id": 1,
        "name": "test",
        "username": "test",
        "email": "test",
    }

    assert True


def test_get_employees(employee_service):
    res = employee_service.get_all_employees()
    assert res == [
        {
            "email": "test",
            "id": 1,
            "name": "test",
            "username": "test"
        }
    ]
import pytest
from unittest.mock import MagicMock
from ..employee_service import EmployeeService
from ...controllers.dtos.employee_controller_dto import CreateEmployeeDTO


@pytest.fixture
def employee_service():
    mock_employee_repository = MagicMock()

    mock_employee_repository.save.return_value = {
        "id": 1,
        "name": "test",
        "username": "test",
        "email": "test",
    }
    mock_employee_repository.get_all_employees.return_value = [
        {
            "email": "test",
            "id": 1,
            "name": "test",
            "username": "test"
        }
    ]

    employee_service = EmployeeService(mock_employee_repository)

    return employee_service


def test_create_employee(employee_service):
    create_employee_mock_input = CreateEmployeeDTO(
        name="test",
        username="test",
        email="test",
    )
    res = employee_service.create_employee(create_employee_mock_input)

    assert res == {
        "id": 1,
        "name": "test",
        "username": "test",
        "email": "test",
    }

    assert True


def test_get_employees(employee_service):
    res = employee_service.get_all_employees()
    assert res == [
        {
            "email": "test",
            "id": 1,
            "name": "test",
            "username": "test"
        }
    ]

Enter fullscreen mode Exit fullscreen mode

原文链接：Implementing Dependency Injection in Python Flask Using Dependency Injector