[Python] SqlAlchemy polymorphic with Pydantic TypeAdapter

I have the next models:

class A(Base):
some_attr: Mapped[str]

class B(A):
problematic_attr: Mapped[CustomPydanticModel | None] = mapped_column(PydanticType(CustomPydanticModel), use_existing_column=True)

class C(A):
problematic_attr: Mapped[DifferentCustomPydanticModel | None] = mapped_column(PydanticType(DifferentCustomPydanticModel ), use_existing_column=True)


the PydanticType is a sqlalchemy TypeDecorator which parse from and to JSON:

class PydanticType(TypeDecorator):
impl = JSON

def __init__(self, pydantic_type):
super().__init__()
self.pydantic_type = TypeAdapter(pydantic_type)


def process_bind_param(self, value, dialect):
return self.pydantic_type_adapter.dump_python(value, mode="json", exclude_unset=True) if value else None

def process_result_value(self, value, dialect):
result = loads(value) if isinstance(value, str) else value
return self.pydantic_type_adapter.validate_python(result, from_attributes=True) if result else None


The problem is, when PydanticType is parsing C().problematic_attr, in self.pydantic_type we have type CustomPydanticModel instead of DifferentCustomPydanticModel.

I know this problem is because of the use_existing_column=True, but I don't know how to fix it. In the DB this is handle using the single table inheritance strategy so I need the use_existing_column

Continue reading...