如何在Python 3.12+中正确使用泛型来提高代码质量？

我正在尝试使用泛型来改进 FastAPI 应用程序中的类型注释。我有一个抽象存储库类，在其中使用泛型：

from abc import ABC, abstractmethod
from typing import List


class AbstractRepository[T](ABC):
    @abstractmethod
    async def add_one(self, data: dict) -> T:
        ...

    @abstractmethod
    async def delete(self, id: int) -> T:
        ...

    @abstractmethod
    async def update(self, id: int, data: dict) -> T:
        ...

    @abstractmethod
    async def get_one(self, id: int) -> T | None:
        ...

    @abstractmethod
    async def get_all(self) -> List[T]:
        ...

名为 SQLAlchemyRepository 的类继承自该抽象类。在此类中，我将泛型绑定到 Base ，这是所有 SQLAlchemy 模型继承自的类：

from typing import Type, List

from sqlalchemy import insert, delete, update, select

from src.database import Database
from src.models import Base
from src.repositories import AbstractRepository


class SQLAlchemyRepository(AbstractRepository[Base]):
    _model: Type[Base] = None

    def __init__(self):
        self.__database: Database = Database()

    async def add_one(self, data: dict) -> Base:
        async with self.__database.get_async_session() as session:
            stmt = insert(self._model).values(**data).returning(self._model)
            res = await session.execute(stmt)
            await session.commit()
            return res.scalar_one()

    async def delete(self, id: int) -> Base:
        async with self.__database.get_async_session() as session:
            stmt = delete(self._model).filter_by(id=id).returning(self._model)
            res = await session.execute(stmt)
            await session.commit()
            return res.scalar_one()

    async def update(self, id: int, data: dict) -> Base:
        async with self.__database.get_async_session() as session:
            stmt = update(self._model).values(**data).filter_by(id=id).returning(self._model)
            res = await session.execute(stmt)
            return res.scalar_one()

    async def get_one(self, id: int) -> Base | None:
        async with self.__database.get_async_session() as session:
            stmt = select(self._model).filter_by(id=id)
            res = await session.execute(stmt)
            res = res.scalar_one_or_none()
            return res

    async def get_all(self) -> List[Base]:
        async with self.__database.get_async_session() as session:
            stmt = select(self._model)
            res = await session.execute(stmt)
            res = res.scalars().all()
            return res

特定 SQLAlchemy 模型的存储库从此存储库继承：

from . import SQLAlchemyRepository
from src.models import Users


class UsersRepository(SQLAlchemyRepository):
    _model = Users

我是否正确使用泛型？我阅读了 Python 3.12 泛型的文档，但找不到任何使用泛型的真实企业示例。

你对于泛型的使用总体来说是 正确 的，并且展现了对泛型概念的良好理解。你成功地使用泛型创建了一个抽象的存储库类，并将其应用于具体的SQLAlchemyRepository和UsersRepository。

以下是一些额外的建议和改进：

1. 泛型类型约束:

• 虽然你的代码可以正常工作，但你可以通过添加类型约束来进一步提高代码的类型安全性。例如，在 AbstractRepository 中， add_one , update 方法中的 data: dict 可以使用 TypedDict 来更精确地定义字典的结构，并与具体的模型属性对应。

```python from typing import TypedDict

class UserCreateData(TypedDict): name: str email: str

class AbstractRepository T : @abstractmethod async def add_one(self, data: UserCreateData) -> T: ... ```

2. 类型推断:

• 在 SQLAlchemyRepository 中，你不需要显式地将 _model 的类型注释为 Type[Base] ，因为类型推断可以自动识别它的类型。

```python class SQLAlchemyRepository(AbstractRepository[Base]): _model = None # 类型将被自动推断为 Type[Base]

```

3. 依赖注入:

• 为了提高代码的可测试性和可维护性，可以考虑使用依赖注入来管理 Database 实例，而不是在 SQLAlchemyRepository 的构造函数中直接实例化它。

python class SQLAlchemyRepository(AbstractRepository[Base]): def __init__(self, database: Database): self.__database = database

4. 错误处理:

• 在真实的企业级应用中，你需要考虑添加适当的错误处理机制。例如，当数据库操作失败时，你应该捕获异常并返回有意义的错误信息。

python async def get_one(self, id: int) -> Base | None: async with self.__database.get_async_session() as session: try: stmt = select(self._model).filter_by(id=id) res = await session.execute(stmt) return res.scalar_one_or_none() except SQLAlchemyError as e: # 处理数据库错误 raise HTTPException(status_code=500, detail=str(e))

总结:

你的代码已经很好地使用了泛型，并展现了对类型提示的良好理解。通过添加类型约束，使用类型推断，采用依赖注入和实现更完善的错误处理，你可以进一步提高代码的质量和可维护性，使其更接近真实的企业级应用标准。