多层架构中的依赖倒置

Question

我在学习FastAPI时一直在学习多层（洋葱）架构。当我开始练习真实的例子时，我遇到了一些困难。请看下面的代码：

from abc import ABC, abstractmethod

from pydantic import BaseModel
from pymongo.collection import Collection
from sqlalchemy.orm import Session


class ProductDto(BaseModel):
    title: str
    weight: float


class Repository(ABC):
    # interface for crud operations

    @abstractmethod
    def get(self, id_): pass

    @abstractmethod
    def create(self, create_data: dict): pass

    # another crud methods


class SqlRepository(Repository):
    # implementation of crud methods in sql (sqlalchemy)

    model = None  # sqlalchemy model

    def __init__(self, session: Session):
        self._session = session

    def get(self, id_):
        print(f'getting instance id={id_} in SqlRepository')

    def create(self, create_data: dict):
        print(f'creating instance in SqlRepository')


class MongoRepository(Repository):
    # implementation of crud methods with mongo

    def __init__(self, collection: Collection):
        self._collection = collection

    def get(self, id_):
        print(f'getting instance id={id_} in MongoRepository')

    def create(self, create_data: dict):
        print(f'creating instance in MongoRepository')


class ProductRepository(Repository):
    # interface for product repository which will be use in type hints
    # method get_random_product was added to basic crud methods

    @abstractmethod
    def get_random_product(self): pass


class ProductSqlRepository(SqlRepository, ProductRepository):
    # sql implementation of ProductRepository interface
    def get_random_product(self):
        print('getting random product in ProductSqlRepository')


class ProductMongoRepository(MongoRepository, ProductRepository):
    # mongo implementation of ProductRepository interface
    def get_random_product(self):
        print('getting random product in ProductMongoRepository')


def process(repository: ProductRepository):
    repository.get(1)


sql_repo = ProductSqlRepository('session')
process(sql_repo)

mongo_repo = ProductMongoRepository('collection')
process(mongo_repo)

我想提供使用不同存储（如 sql、mongo 等）的可能性，这样我就可以注入其他存储库。

看起来还好吗？我对这个继承表示怀疑：

class ProductMongoRepository(MongoRepository, ProductRepository):
    def get_random_product(self):
        print('getting random product in ProductMongoRepository')

如有需要请更正。

Answer 1

您的方法是有效的，但我认为在这种情况下：“组合优于继承”适用。 正如您提到的，这种继承可能有点令人困惑：

class ProductMongoRepository(MongoRepository, ProductRepository):
    def get_random_product(self):
        print('getting random product in ProductMongoRepository')

那么，我更愿意做作文：

class ProductRepository(Repository):
    # interface for product repository which will be use in type hints
    # method get_random_product was added to basic crud methods

    @abstractmethod
    def get_random_product(self): pass


class ProductSqlRepository(ProductRepository):
    # sql implementation of ProductRepository interface
    def __init__(self, base_repository = SqlRepository("session")):
        self.base_repository = base_repository

    def get_random_product(self):
        print('getting random product in ProductSqlRepository')


class ProductMongoRepository(ProductRepository):
    # mongo implementation of ProductRepository interface
    def __init__(self, base_repository = MongoRepository("collection")):
        self.base_repository = base_repository

    def get_random_product(self):
        print('getting random product in ProductMongoRepository')


def process(repository: ProductRepository):
    repository.get(1)


sql_repo = ProductSqlRepository()
process(sql_repo)

mongo_repo = ProductMongoRepository()
process(mongo_repo)

您可能需要调整一些事情，但一般来说，想法是不是继承，而是将类的实例作为属性（依赖注入）。请注意，在本例中，我们注入了一个看似基本 CRUD 方法的实例（实际上，该接口可能称为 ICRUDRepository）。但也许您还想为另一组相关方法定义另一个接口，您也可以轻松注入该依赖项。