Fix: webhook parsing when having nested union types

[frankie567]

This is an attempt at solving #50.

As discussed there, we can leverage the new callable Discriminator feature from Pydantic 2 (released in 2.5.0).

---

What we do here is to detect the nested unions in the webhook schemas. For those ones, we add a specific handling:

1. Add a Tag annotation on each individual schema. This allows us to identify a specific schema by a name in the discriminator function
2. Add SchemaDiscriminator annotation on the nested union schema.

This SchemaDiscriminator is a custom function implemented in webhooks/discriminator.py that returns us the Pydantic Discriminator with a dynamic discriminator function (aka a factory).

Indeed, the only way I found to determine which is the right schema to apply in the union is to actually check for all the required fields of the schema. The first one who has all the required fields wins. That's why we pass again the whole list of schema to generate this function.

---

It solves the problem, but the implementation is maybe a bit obscure. Open to discussion about how we could improve this.

Cheers!

[yanyongyu]

I found another solution:

make a fake model schema just works 😂

from typing import Any, Union, Literal, Annotated

from pydantic_core import core_schema
from pydantic import Field, BaseModel, TypeAdapter, GetCoreSchemaHandler

class UnionSchema:
    def __init__(self, type: Any, tag: str) -> None:
        self.type = type
        self.tag = tag

    def parse_obj(self, obj: Any) -> Any:
        return TypeAdapter(self.type).validate_python(obj)

    def __get_pydantic_core_schema__(
        self, _source_type: Any, _handler: GetCoreSchemaHandler
    ) -> core_schema.CoreSchema:
        return core_schema.no_info_before_validator_function(
            self.parse_obj,
            core_schema.model_schema(
                BaseModel,
                schema=core_schema.model_fields_schema(
                    {
                        "type": core_schema.model_field(
                            core_schema.literal_schema([self.tag])
                        )
                    }
                ),
            ),
        )

class A(BaseModel):
    type: Literal["a"] = "a"

class BOneOf0(BaseModel):
    type: Literal["b"] = "b"
    x: int

class BOneOf1(BaseModel):
    type: Literal["b"] = "b"
    y: int

MyType = Annotated[
    Union[A, Annotated[Any, UnionSchema(Union[BOneOf0, BOneOf1], "b")]],
    Field(discriminator="type"),
]

print(repr(TypeAdapter(MyType).validate_python({"type": "b", "y": 1})))

[frankie567]

Sorry for long delay 😊

I'm not sure to understand why your solution works but looks like a good one 😄 What do you prefer? Shall we pursue with my original proposal or do we go for the UnionSchema trick?

[yanyongyu]

My solution is to create a fake model schema for the union type. The fake model schema contains only one field, the discriminator field. Asign the Litereal type to the discriminator field of the fake model schema will make pydantic to known the union type is a set of type with same discriminator choice. The example above is just a test code, we can make it more convenient.

[frankie567]

Hmm, I see. But how does Pydantic detects that it's BOneOf0 or BOneOf1? Is it because when calling parse_obj method, one of the type throws a validation error?

Anyway, I can explore this, if you're okay :)

[yanyongyu]

ways of Pydantic detecting union types is described here: https://docs.pydantic.dev/latest/concepts/unions/#union-modes

The "left to right" mode is the same with Pydantic v1.

[yanyongyu]

@frankie567 I have changed this pr branch in my local repo but i do not have the permission to push to the fix/50 branch. is there any way to update this pr? :joy: i just fixed this parsing error with the approach mentioned above.

May be i could push to this repo and sync to the fix/50 branch of polarsource fork?

[frankie567]

Damned, TIL that maintainers can only push to a PR from a fork that's user-owned (?!).

I added you as member of our fork, so you should be able to push to this PR.

[yanyongyu]

I will merge this, add some test cases and then publish a pre-release for the features just added.