課題
基本的に v1/v2 のモデルを混合して使うことはできないのですが、Pydantic v1 → v2 のマイグレーションの都合上どうしても v1/v2 を共存させる必要がありました。
例えば以下のように v2 モデルの中に v1 モデルのフィールドを作るとエラーが発生します。
from pydantic import BaseModel
from pydantic.v1 import BaseModel as BaseModelV1
class Child(BaseModelV1):
name: str
class Parent(BaseModel):
child: Child
if __name__ == "__main__":
print("Parent schema :", Parent.model_json_schema())
print("Parent instance:", parent := Parent.model_validate({"child": {"name": "John"}}))
print("Parent dumped :", parent.model_dump())
...
raise PydanticInvalidForJsonSchema(f'Cannot generate a JsonSchema for {error_info}')
pydantic.errors.PydanticInvalidForJsonSchema: Cannot generate a JsonSchema for core_schema.PlainValidatorFunctionSchema ({'type': 'with-info', 'function': <bound method BaseModel.validate of <class '__main__.Child'>>})
For further information visit https://errors.pydantic.dev/2.10/u/invalid-for-json-schema
v1 と v2 では validation/serialization の挙動が大きく変わっていることもあり、公式にはこれらを併用する手段が提供されていません。
解決策
Pydantic v2 ではモデルに __get_pydantic_core_schema__/__get_pydantic_json_schema__ メソッドを実装することで独自のスキーマ情報を渡すことができます1。
今回は Pydantic v1 のモデルにこれらのメソッドを追加して併用可能にすることにします。
例えば以下のような関数を実装します。
from typing import Any, TypeVar
from pydantic import BaseModel, GetCoreSchemaHandler, GetJsonSchemaHandler
from pydantic.json_schema import JsonSchemaValue
from pydantic.v1 import BaseModel as BaseModelV1
from pydantic_core import core_schema
_TV1 = TypeVar("_TV1", bound=BaseModelV1 | type[BaseModelV1])
def _get_pydantic_core_schema(
cls: type[BaseModelV1],
source_type: Any,
handler: GetCoreSchemaHandler,
) -> core_schema.CoreSchema:
del source_type, handler
# NOTE: ちゃんと実装すれば v1 モデルから適切な CoreSchema を生成できそうな気がしますが、
# 非常に複雑になるためひとまず AnySchema を返しています。
return core_schema.no_info_before_validator_function(
function=lambda value: value if isinstance(value, cls) else cls.parse_obj(value),
schema=core_schema.any_schema(),
)
def _get_pydantic_json_schema(
cls: type[BaseModelV1],
source_type: Any,
handler: GetJsonSchemaHandler,
) -> JsonSchemaValue:
del source_type, handler
return cls.schema()
def pydantic_v2_compatible(model: _TV1) -> _TV1:
given = model
cls = given.__class__ if isinstance(given, BaseModel) else given
setattr(cls, "__get_pydantic_core_schema__", classmethod(_get_pydantic_core_schema))
setattr(cls, "__get_pydantic_json_schema__", classmethod(_get_pydantic_json_schema))
return given
...
@pydantic_v2_compatible
class Child(BaseModelV1):
name: str
これを使って先述のコードを再度実行します。
Parent schema : {'properties': {'child': {'properties': {'name': {'title': 'Name', 'type': 'string'}}, 'required': ['name'], 'title': 'Child', 'type': 'object'}}, 'required': ['child'], 'title': 'Parent', 'type': 'object'}
Parent instance: child=Child(name='John')
Parent dumped : {'child': Child(name='John')}
JSON Schema の生成と validation はうまく動作しています。
model_dump の結果を辞書型にする場合は serializer を設定する必要があります2。
def serialize_model(model: BaseModelV1, info: core_schema.SerializationInfo) -> dict[str, Any]:
return model.dict(
include=info.include,
exclude=info.exclude,
by_alias=info.by_alias,
exclude_defaults=info.exclude_defaults,
exclude_none=info.exclude_none,
exclude_unset=info.exclude_unset,
)
class Parent(BaseModel):
child: Annotated[Child, PlainSerializer(serialize_model)]
Parent schema : {'properties': {'child': {'properties': {'name': {'title': 'Name', 'type': 'string'}, 'age': {'title': 'Age', 'type': 'integer'}}, 'required': ['name'], 'title': 'Child', 'type': 'object'}}, 'required': ['child'], 'title': 'Parent', 'type': 'object'}
Parent instance: child=Child(name='John')
Parent dumped : {'child': {'name': 'John'}}
おわりに
用途によっては他にも変更が必要かもしれませんが、これでひとまず基本的なユースケースでは問題なく動作するようになりました。
追記 (2025-06-07)
CoreSchema を設定する際に serialization 引数を通して serialization の挙動を設定できます。
return core_schema.no_info_after_validator_function(
lambda v: v if isinstance(v, cls) else cls.parse_obj(v),
core_schema.any_schema(
serialization=core_schema.plain_serializer_function_ser_schema(
function=_serialize_model,
info_arg=True,
)
),
)