Dependency injection framework

https://python-dependency-injector.ets-labs.org/index.html

Dependency Injector is a dependency injection framework for Python.

It helps implementing the dependency injection principle.

Key features of the Dependency Injector:

• Providers. Provides Factory, Singleton, Callable, Coroutine, Object, List, Dict, Configuration, Resource, Dependency, and Selector providers that help assemble your objects. See Providers.

• Overriding. Can override any provider by another provider on the fly. This helps in testing and configuring dev/stage environment to replace API clients with stubs etc. See Provider overriding.

Decoupled packages example (multiple containers)

https://python-dependency-injector.ets-labs.org/examples/decoupled-packages.html

This example shows how to use Dependency Injector to create decoupled packages.

To achieve a decoupling each package has a container with the components. When a component needs a dependency from the outside of the package scope we use the Dependency provider. The package container has no knowledge on where the dependencies come from. It states a need that the dependencies must be provided. This helps to decouple a package from the 3rd party dependencies and other packages.

To wire the packages we use an application container. Application container has all 3rd party dependencies and package containers. It wires the packages and dependencies to create a complete application.

We build an example micro application that consists of 3 packages:

• user - a package with user domain logic, depends on a database

• photo - a package with photo domain logic, depends on a database and AWS S3

• analytics - a package with analytics domain logic, depends on the user and photo package components

DIP abstraction layer to improve

问题

上面例子将每个对象（user photo analytics）设置为一个独立包， 每个包中有对应一个container

从业务逻辑上，对于上层包 analytics 依赖底层包 user 和 photo 的

本例子中， 对于analytics包的依赖做了声明， 仅仅是两个依赖（user_repositories photo_repositories）， 但是具体的依赖对象的规格（API）是完全不清楚的

analytics包内是无法调用底层包（user photo）的接口的， 例如在 AggregationService 中调用user和photo接口做数据统计，

所以这里的例子仅仅是依赖性注入展示， 无法应用到实际上项目中。

Listing of the example/analytics/containers.py:

"""Analytics containers module."""

from dependency_injector import containers, providers

from . import services


class AnalyticsContainer(containers.DeclarativeContainer):

    user_repository = providers.Dependency()
    photo_repository = providers.Dependency()

    aggregation_service = providers.Singleton(
        services.AggregationService,
        user_repository=user_repository,
        photo_repository=photo_repository,
    )

改进

对于上下层之间解耦， SOLID设计原则有对应的解法， DIP依赖反转原则，

即上层对象不直接生成和调用下层对象

下层对应对象定义 抽象接口，

上层对象只依赖 抽象接口， 这个接口是上下层之间的合约，供双方遵守，

下层对象按照 抽象接口 实现其自身。

https://fanqingsong.github.io/fastapi-hive/design/

 

DIP Definition

* High-level cornerstones should not depend on low-level cornerstones. 
    Both should depend on the abstraction.

* Abstractions should not depend on details. 
    Details should depend on abstractions.

https://github.com/fanqingsong/python-dependency-injector/tree/master/examples/miniapps/decoupled-packages/example

analytics service abstraction

"""Analytics services module."""

import abc


from ..photo.repositories import PhotoRepositoryMeta
from ..user.repositories import UserRepositoryMeta


class AggregationServiceMeta(metaclass=abc.ABCMeta):

    def __init__(self, user_repository: UserRepositoryMeta, photo_repository: PhotoRepositoryMeta):
        self.user_repository: UserRepositoryMeta = user_repository
        self.photo_repository: PhotoRepositoryMeta = photo_repository

    @abc.abstractmethod
    def call_user_photo(self):
        """Must be implemented in order to instantiate."""
        pass

analytics service implementation

"""Analytics services module."""


from ..abstraction.analytics.services import AggregationServiceMeta
from ..abstraction.photo.repositories import PhotoRepositoryMeta
from ..abstraction.user.repositories import UserRepositoryMeta


class AggregationService(AggregationServiceMeta):

    def __init__(self, user_repository: UserRepositoryMeta, photo_repository: PhotoRepositoryMeta):
        self.user_repository: UserRepositoryMeta = user_repository
        self.photo_repository: PhotoRepositoryMeta = photo_repository

    def call_user_photo(self):
        user1 = self.user_repository.get(id=1)
        user1_photos = self.photo_repository.get_photos(user1.id)
        print(f"Retrieve user id={user1.id}, photos count={len(user1_photos)} from aggregation service.")

__main__.py (顶层业务实现)

"""Main module."""

from dependency_injector.wiring import Provide, inject

from .abstraction.user.repositories import UserRepositoryMeta
from .abstraction.photo.repositories import PhotoRepositoryMeta
from .abstraction.analytics.services import AggregationServiceMeta

from .containers import ApplicationContainer


@inject
def main(
        user_repository: UserRepositoryMeta = Provide[
            ApplicationContainer.user_package.user_repository
        ],
        photo_repository: PhotoRepositoryMeta = Provide[
            ApplicationContainer.photo_package.photo_repository
        ],
        aggregation_service: AggregationServiceMeta = Provide[
            ApplicationContainer.analytics_package.aggregation_service
        ],
) -> None:
    user1 = user_repository.get(id=1)
    user1_photos = photo_repository.get_photos(user1.id)
    print(f"Retrieve user id={user1.id}, photos count={len(user1_photos)}")

    user2 = user_repository.get(id=2)
    user2_photos = photo_repository.get_photos(user2.id)
    print(f"Retrieve user id={user2.id}, photos count={len(user2_photos)}")

    assert aggregation_service.user_repository is user_repository
    assert aggregation_service.photo_repository is photo_repository
    print("Aggregate analytics from user and photo packages")

    aggregation_service.call_user_photo()


if __name__ == "__main__":
    application = ApplicationContainer()
    application.wire(modules=[__name__])

    main()

containers.py (顶层对应组装)

上面三个都是面向抽象接口编程，

具体实例化组装的工作在这里实现。

"""Containers module."""

import sqlite3

import boto3
from dependency_injector import containers, providers

from .user.containers import UserContainer
from .photo.containers import PhotoContainer
from .analytics.containers import AnalyticsContainer


class ApplicationContainer(containers.DeclarativeContainer):

    config = providers.Configuration(ini_files=["config.ini"])

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

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

    user_package = providers.Container(
        UserContainer,
        database=sqlite,
    )

    photo_package = providers.Container(
        PhotoContainer,
        database=sqlite,
        file_storage=s3,
    )

    analytics_package = providers.Container(
        AnalyticsContainer,
        user_repository=user_package.user_repository,
        photo_repository=photo_package.photo_repository,
    )