我正在实现一个存储库模式作为打字练习。
我有几个不相关的 SQLAlchemy 模型:
class Base(MappedAsDataclass, DeclarativeBase):
id: Mapped[primary_key] = mapped_column(init=False)
default_string = Annotated[str, mapped_column(String(100))]
class User(Base):
__tablename__ = "sample"
name: Mapped[default_string]
class Sample(Base):
__tablename__ = "sample"
value: Mapped[default_string]
location: Mapped[default_string]
我想为每个创建一个
Repository
,以便有统一的交互和查询方式,并且我想使用打字来提示用法。
说:
T = TypeVar("T", bound=Base)
class Repository(ABC, Generic[T]):
def __init__(self, model: type[T]):
self.db = SessionLocal()
self.model = model
def list(self, skip: int = 0, limit: int = 100):
return self.db.query(self.model).offset(skip).limit(limit).all()
所以我将它们用作:
class SampleRepository(Repository[Sample]):
def get_by_name(self, name: str) -> list[Sample]:
return self.model.query.filter(name=name).all()
repository = SampleRepository(Sample)
现在,事情是,在
repository.create()
上,这是一个相当复杂的逻辑,我需要为不同的模型参数拥有不同的签名。我想到在这种情况下使用字典解包
class Repository(ABC, Generic[T]):
...
def create(self, **data) -> T:
instance = self.model(**data)
self.db.add(instance)
self.db.commit()
self.db.refresh(instance)
return instance
但是如果我尝试在
SampleRepository
中重载此方法:
class SampleRepository(Repository[Sample]):
def create(self, name: str) -> Sample:
return super().create(
name=name,
)
这会中断,因为“create”的签名与超类型“Repository”不兼容。
有办法实现这一点吗?或者我对类型系统要求太多?
你就快到了!对于这样的场景,Python 中的类型系统可能有点棘手,因为它本身不支持子类中具有不同方法签名的“部分特定”覆盖。但是,您可以通过使用
Protocol
和 TypedDict
技术为每个模型的参数提供结构化类型来实现这种灵活性,从而允许 create
方法根据 Repository
子类型具有不同的签名。
为每个模型的
create
参数定义一个TypedDict。
TypedDict
指定创建每个模型所需的字段。这样,每个存储库都会有一个唯一的 create
签名。实现
Protocol
用于结构化类型。
Protocol
约束每个模型的必填字段并强制输入。调整
Repository
以接受TypedDict
在create
中拆包。
create
方法可以使用这些TypedDict
类型来满足特定的模型数据要求。此方法如下所示:
from typing import Protocol, TypeVar, Generic, TypedDict, Type
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import DeclarativeMeta
from abc import ABC
# Base class setup
class Base:
id: int
# Model-specific TypedDict for `create` parameters
class SampleData(TypedDict):
value: str
location: str
class UserData(TypedDict):
name: str
# Protocol defining `create` parameters per model type
class ModelData(Protocol):
@classmethod
def __init_subclass__(cls, **kwargs):
...
T = TypeVar("T", bound=Base)
# Repository base class with a generic `create` method
class Repository(ABC, Generic[T]):
def __init__(self, model: Type[T], db: Session):
self.model = model
self.db = db
def list(self, skip: int = 0, limit: int = 100):
return self.db.query(self.model).offset(skip).limit(limit).all()
def create(self, **data: ModelData) -> T:
instance = self.model(**data)
self.db.add(instance)
self.db.commit()
self.db.refresh(instance)
return instance
# Sample repository inheriting the Repository and specifying `create`'s parameters
class SampleRepository(Repository[Sample]):
def create(self, value: str, location: str) -> Sample:
return super().create(value=value, location=location)
# User repository for `User` model
class UserRepository(Repository[User]):
def create(self, name: str) -> User:
return super().create(name=name)
# Usage example
db_session = SessionLocal()
sample_repo = SampleRepository(Sample, db_session)
sample_instance = sample_repo.create(value="some_value", location="some_location")
这里有一些解释
SampleData
、UserData
)定义每个模型的 create
参数的结构,允许您在创建实例时区分 Sample
和 User
。ModelData
) 允许每个属性字典动态满足类型要求。create
签名,而不会破坏与 Repository
超类的兼容性,从而在打字系统中实现所需的灵活性。