Bring dmod into compliance with pydantic v2

[aaraney]

pydantic v2 also ships with pydantic v1, so in the short term, imports just need to be updated to pull from pydantic.v1 instead of pydantic. It will be a much larger effort to become compliant using the v2 apis. This is due to a large number of api changes.

[robertbartel]

After a brief discussion this morning (and @aaraney, correct me if I misstate anything), it seems the solution of simply continuing to use the Pydantic V1 API provided by the V2 package may not suffice. A V1 model will not work with a field that is a V2 model. In particular, if NOAA-OWP/ngen-cal#119 requires any models to fully upgrade to V2, and any such models are used as fields for DMOD models, then some or all of DMOD's Pydantic usage will have to be fully migrated to V2.

@hellkite500, that means NOAA-OWP/ngen-cal#119 will have impacts on when and how this must be addressed that I think weren't previously expected. We may want to discuss and coordinate further.

[aaraney]

See pydantic/pydantic#7637 and pydantic/pydantic#7997

[robertbartel]

@aaraney, just as a sanity check ...

If I'm following correctly, both of the referenced issues involve a V2 model being incompatible with a field that is a V1 model. That's not the scenario we'd need. The DMOD classes would remain V1 models, so here it's V1 models that may need V2 fields.

I've read through those issues and some of the Pydantic docs quickly, and I didn't immediately see anything clearly confirming our scenario is also broken. Have you already encountered something that demonstrates incompatibility with things this direction?

[aaraney]

I should have included some runnable code that demonstrates the problem! As you stated, embedding a V1 model in a V2 model is not our use case, but regardless of the composition order there still is an issue:

import pydantic as v2
import pydantic.v1 as v1


def test_v1_in_a_v2():
    class Foo(v1.BaseModel): ...

    class Bar(v2.BaseModel):
        foo: Foo

    Bar(foo=Foo())
    # TypeError: validate() takes 2 positional arguments but 3 were given


def test_v2_in_a_v1():
    class Baz(v2.BaseModel): ...

    class Quox(v1.BaseModel):
        baz: Baz

    Quox(baz=Baz())
    # venv/lib/python3.9/site-packages/pydantic/v1/main.py:197: in __new__
    #     fields[ann_name] = ModelField.infer(
    # venv/lib/python3.9/site-packages/pydantic/v1/fields.py:504: in infer
    #     return cls(
    # venv/lib/python3.9/site-packages/pydantic/v1/fields.py:434: in __init__
    #     self.prepare()
    # venv/lib/python3.9/site-packages/pydantic/v1/fields.py:555: in prepare
    #     self.populate_validators()
    # venv/lib/python3.9/site-packages/pydantic/v1/fields.py:829: in populate_validators
    #     *(get_validators() if get_validators else list(find_validators(self.type_, self.model_config))),

[aaraney]

Just to update this conversation, I may have found a way forward in the short term. The following code enables embedding a v2 model in a v1 model:

from __future__ import annotations
from typing import Any, Optional, Union

import pydantic as v2
import pydantic.v1 as v1


class Base(v1.BaseModel):
    @classmethod
    def _get_value(
        cls,
        v: Any,
        to_dict: bool,
        by_alias: bool,
        include: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]],
        exclude: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]],
        exclude_unset: bool,
        exclude_defaults: bool,
        exclude_none: bool,
    ) -> Any:
        if isinstance(v, v2.BaseModel):
            return v.model_dump(
                by_alias=by_alias,
                include=include,
                exclude=exclude,
                exclude_unset=exclude_unset,
                exclude_defaults=exclude_defaults,
                exclude_none=exclude_none,
            )
        return v1.BaseModel._get_value(
            v=v,
            to_dict=to_dict,
            by_alias=by_alias,
            include=include,
            exclude=exclude,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
        )


class V2(v2.BaseModel):
    field: int

    @classmethod
    def __get_validators__(cls) -> CallableGenerator:
        def validate(values):
            return cls.model_validate(values)

        yield validate


class V1(Base):
    v2: V2


def test_it_two_in_one():
    o = V1(v2=V2(field=42))

    assert V1.parse_obj({"v2": V2(field=42)}) == o
    assert V1.parse_obj({"v2": {"field": 42}}) == o
    assert V1.parse_raw('{"v2": {"field": 42}}') == o

    assert o.dict() == {"v2": {"field": 42}}
    assert o.json() == '{"v2": {"field": 42}}'

To note, v1.BaseModel.schema_json() does not work with this. I've not looked into how to get that to work.

The above examples use inheritance to achieve this, but there is nothing stopping us from monkey patching v1.BaseModel and v2.BaseModel to achieve the same.

[aaraney]

You can embed a v1 in a v2 and generate a schema doing something like this:

class Foo(v1.BaseModel):

    @classmethod
    def __get_pydantic_json_schema__(
        cls,
        _core_schema: CoreSchema,
        handler: v2.GetJsonSchemaHandler,
        /,
    ) -> JsonSchemaValue:
        return cls.schema()

However, this is not our use case.

[aaraney]

v1 generates schema's using a depth first search over the fields of a model. In part because of this, I've not determined a place we could inject custom logic to handle generating v2 schemas embedded in a v1 model.

[aaraney]

Lots of magic to get this working in a generic sense, but I think ive got it figured out:

def __modify_schema__(
    cls: v2.BaseModel, field_schema: Dict[str, Any], field: Optional[ModelField]
):
    field_schema.update(cls.model_json_schema())

from pydantic._internal._model_construction import ModelMetaclass
new = ModelMetaclass.__new__
def __new__(
    mcs,
    cls_name: str,
    bases: tuple[type[Any], ...],
    namespace: dict[str, Any],
    __pydantic_generic_metadata__: PydanticGenericMetadata | None = None,
    __pydantic_reset_parent_namespace__: bool = True,
    _create_model_module: str | None = None,
    **kwargs: Any,
) -> type:
    cls = new(
        mcs,
        cls_name,
        bases,
        namespace,
        __pydantic_generic_metadata__,
        __pydantic_reset_parent_namespace__,
        _create_model_module,
        **kwargs,
    )
    cls.__modify_schema__ = classmethod(__modify_schema__)
    return cls
ModelMetaclass.__new__ = __new__

class A(v2.BaseModel):
    field: int

    @classmethod
    def __get_validators__(cls) -> CallableGenerator:
        def validate(values):
            return cls.model_validate(values)

        yield validate

class B(v1.BaseModel):
    a: A

print(B.schema_json())

[aaraney]

v2's metaclass, ModelMetadata, that builds a model, checks for the presence of __modify_schema__ on the incoming bases and throws an exception if it is present. The above gets around this.