我定义了一个名为 MetaPersistent 的类:
class MetaPersistent(type(BaseModel)):
Ref: ClassVar[type[RefBase[Self]]]
Repository: ClassVar[type[RepositoryBase[Self]]]
def __new__(cls, *args, skip_meta_init=False, **kwargs):
return super().__new__(cls, *args, **kwargs)
def __init__(self, *args, skip_meta_init=False, **kwargs):
super().__init__(*args, **kwargs)
# name, bases, dict = args
if skip_meta_init is True:
return
# Make a repository class for the new persistent class
rep_bases = tuple(
{
getattr(base, "Repository"): None
for base in self.__mro__
if hasattr(base, "Repository")
}.keys()
)
rep_cls = type(
"Repository",
rep_bases,
{
"__module__": self.__module__,
"__qualname__": f"{self.__qualname__}.Repository",
"__persistent__": self,
},
)
# Instance the repository
repository =
rep_cls(_english_inflector.tableize(self.__qualname__))
# Make a reference class for the new persistent class
ref_bases = tuple(
{
getattr(base, "Ref"): None
for base in self.__mro__
if hasattr(base, "Ref")
}.keys()
)
ref_cls = type(
"Ref",
ref_bases,
{
"__module__": self.__module__,
"__qualname__": f"{self.__qualname__}.Ref",
"__persistent__": self,
"__repository__": repository,
},
)
setattr(self, "Repository", rep_cls)
setattr(self, "Ref", ref_cls)
setattr(self, "__repository__", repository)
从中继承持久化:
class Persistent(BaseModel, metaclass=MetaPersistent, skip_meta_init=True):
id: Optional[Id] = Field(None, title="Unique object Id", alias="_id")
@property
def ref(self) -> RefBase[Self]:
return self.Ref(self.id)
async def store(self) -> RefBase[Self]:
if self.id:
await repository_of(self).update_obj(self)
else:
await repository_of(self).insert_obj(self)
return self.ref
然后,一个名为Vehicle的类继承自它:
class Vehicle(Persistent):
tenant_ref: Optional[Tenant.Ref] = None # Reference to the tenant (default is None)
loadLimits: float # Load limits of the vehicle
startLocation: Location # Starting location of the vehicle
endLocation: Location # Ending location of the vehicle
class Config:
arbitrary_types_allowed = True # Allows arbitrary types
当我通过以下方式实例化车辆时出现问题:
vehicles = [Vehicle(
loadLimits=vehicle["loadLimits"],
startLocation=Location(**vehicle["startLocation"]),
endLocation=Location(**vehicle["endLocation"])
) for vehicle in json_data_vehicles]
Pydantic投掷:
vehicles = [Vehicle(
^^^^^^^^
line 164, in __init__
__pydantic_self__.__pydantic_validator__.validate_python(data,
self_instance=__pydantic_self__)
TypeError: BaseModel.validate() takes 2 positional arguments but 3 were given
另一方面,如果Vehicle像Vehicle(BaseModel)一样直接继承自BaseModel,则不会发现问题。猜测 Pydantic 验证继承的方式存在一些问题,但我无法修复它。
事实证明,Pydantic v1 和 v2 之间存在一些问题。由于我的一些导入依赖于 v1,代码因未知原因崩溃了。我通过从 v2(默认的 v2)导入所有内容来修复此问题:
from pydantic.v1 import BaseModel #importing from v1. Code crashes
from pydantic import BaseModel #importing from v2. Code runs ok