能否混合抽象基类(ABC)与Pydantic BaseModel?代码报错及实现疑问
问题解答
一、混合ABC与Pydantic BaseModel的设计是否可取?
完全可行。ABC用来定义抽象接口约束子类行为,BaseModel提供数据验证、序列化能力,两者结合既能保证代码规范性,又能利用Pydantic的特性,不存在本质冲突。你当前的问题不是多继承导致的,而是Ticker作为自定义包装类,没有实现Pydantic所需的schema生成逻辑,导致ConcreteMessage无法识别这个字段类型。
二、解决报错并实现需求的两种方案
方案1:用Pydantic Union+验证器简化实现
直接用Union类型接收str/int输入,通过Pydantic验证器转换为对应的Ticker实例,同时保留ABC的抽象约束:
from abc import ABC, abstractmethod from pydantic import BaseModel, field_validator, ValidationError from typing import Union # 抽象接口约束必须实现的方法 class AbstractTicker(ABC): @abstractmethod def to_str(self) -> str: pass # 字符串Ticker实现 class TickerStr(AbstractTicker, BaseModel): value: str @field_validator('value') def check_valid_str(cls, v): if not v: raise ValueError('字符串Ticker不能为空') return v def to_str(self) -> str: return self.value def __str__(self) -> str: return f'Ticker[str]({self.value})' # 整数Ticker实现 class TickerInt(AbstractTicker, BaseModel): value: int @field_validator('value') def check_valid_int(cls, v): if v <= 0: raise ValueError('整数Ticker必须大于0') return v def to_str(self) -> str: return str(self.value) def __str__(self) -> str: return f'Ticker[int]({self.value})' # 消息模型:自动转换输入为对应Ticker实例 class ConcreteMessage(BaseModel): ticker: Union[TickerStr, TickerInt] @field_validator('ticker', mode='before') def convert_to_ticker(cls, v): if isinstance(v, str): return TickerStr(value=v) elif isinstance(v, int): return TickerInt(value=v) raise TypeError(f'不支持的类型:{type(v)},仅接受字符串或整数') def __str__(self) -> str: return str(self.ticker) def main(): # 正常使用:直接传入str/int msg1 = ConcreteMessage(ticker='NVDA') msg2 = ConcreteMessage(ticker=1234) print(msg1) print(msg2) # 验证失败示例(会抛出ValidationError) # try: # msg3 = ConcreteMessage(ticker='') # msg4 = ConcreteMessage(ticker=-5) # except ValidationError as e: # print(e) if __name__ == '__main__': main()
方案2:让原有Ticker类支持Pydantic Schema
如果你想保留原有的Ticker包装类结构,只需给Ticker添加__get_pydantic_core_schema__方法,告诉Pydantic如何处理这个类型:
from abc import ABC, abstractmethod from pydantic import BaseModel, GetCoreSchemaHandler from pydantic_core import core_schema class AbstractTicker(ABC): @abstractmethod def to_str(self) -> str: pass @abstractmethod def _get_value(self) -> str|int: pass class TickerStr(AbstractTicker, BaseModel): _ticker_str: str def __init__(self, ticker_str) -> None: super().__init__() assert isinstance(ticker_str, str), 'ticker必须是字符串类型' assert len(ticker_str) > 0, 'ticker不能为空字符串' self._ticker_str = ticker_str def __str__(self) -> str: return f'Ticker[str]({self._ticker_str})' def _get_value(self) -> str: return self._ticker_str def to_str(self) -> str: return self._ticker_str class TickerInt(AbstractTicker, BaseModel): _ticker_int: int def __init__(self, ticker_int) -> None: super().__init__() assert isinstance(ticker_int, int), 'ticker必须是整数类型' assert ticker_int > 0, 'ticker必须大于0' self._ticker_int = ticker_int def __str__(self) -> str: return f'Ticker[int]({self._ticker_int})' def _get_value(self) -> int: return self._ticker_int def to_str(self) -> str: return str(self._ticker_int) class Ticker: _ticker: AbstractTicker def __init__(self, ticker: object): if isinstance(ticker, str): self._ticker = TickerStr(ticker) elif isinstance(ticker, int): self._ticker = TickerInt(ticker) else: raise TypeError(f'不支持的类型 {type(ticker)} 作为ticker') def __str__(self) -> str: return str(self._ticker) def to_str(self) -> str: return self._ticker.to_str() def __eq__(self, ticker: object) -> bool: if isinstance(ticker, Ticker): return self._ticker._get_value() == ticker._ticker._get_value() return False # 核心:实现Pydantic schema生成逻辑 @classmethod def __get_pydantic_core_schema__(cls, source_type, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: # 定义验证逻辑:将str/int转换为Ticker,或者直接返回已有的Ticker实例 def validate(v): if isinstance(v, (str, int)): return Ticker(v) elif isinstance(v, Ticker): return v raise TypeError('ticker必须是字符串、整数或Ticker实例') # 返回schema,包含验证和序列化规则 return core_schema.no_info_wrap_validator_function( validate, handler.generate_schema(str | int), serialization=core_schema.plain_serializer_function_ser(lambda x: x._ticker._get_value()), ) class ConcreteMessage(BaseModel): ticker: Ticker def __str__(self) -> str: return str(self.ticker) def main(): str_ticker = TickerStr('NVDA') int_ticker = TickerInt(1234) print(str_ticker) print(int_ticker) # 直接传入str,Pydantic自动转换为Ticker concrete_message = ConcreteMessage(ticker='TEST') print(concrete_message) if __name__ == '__main__': main()
关键说明
- 报错根源:Pydantic无法为非BaseModel且未实现schema逻辑的自定义类型生成验证规则,所以要么让类型成为BaseModel子类,要么实现
__get_pydantic_core_schema__。 - 两种方案各有优劣:方案1更简洁,利用Pydantic原生特性;方案2保留了你原有的包装类结构,适合不想大幅修改代码的场景。
内容的提问来源于stack exchange,提问作者user2138149
相关产品推荐
相关产品推荐

