将 sqlalchemy orm 模型（PostgreSQL Ltree 扩展）转换为 pydantic

Question

如何正确地将我的 sqlalchemy ORM 模型字段 [使用 PostgreSQL LTREE 扩展] 转换为 pydantic 模型或字符串 [稍后在 FastAPI 中使用]？

---

我有 sqlalchemy 和 pydantic 模型：

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy_utils import LtreeType  # Postgres LTREE extension

from pydantic import BaseModel

# Sqlalchemy ORM <- Works as expected, fetches data
class ItemORM(Base):
    __tablename__ = 'item'

    id: Mapped[int] = mapped_column(primary_key=True)  # 1
    label: Mapped[str]  # 'Some custom label'
    path: Mapped[str] = mapped_column(LtreeType) # 'A.23', THIS TO BE CONVERTED TO PYDANTIC OR STRING

# Pydantic
class Item(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True) # Allows to use LtreeType type for pydantic

    id: int
    label: str
    path: LtreeType | str

我的 FastAPI 代码用于获取数据并将其显示给用户：

async get_data():
  with Session() as session:
    db_data = session.query(ItemORM).first() # '[(<ItemORM id='1',label='Label',path='A.21'>)]'
    item_dto: Item = Item.model_validate(db_data, from_attributes=True) # <- this generates error

当我尝试执行它时，它给出错误：

item.path.is-instance[LtreeType]
  Input should be an instance of LtreeType [type=is_instance_of, input_value=Ltree('A.23'), input_type=Ltree]
  For further information visit https://errors.pydantic.dev/2.9/v/is_instance_of

item.path.str
  Input should be a valid string [type=string_type, input_value=Ltree('A.23'), input_type=Ltree]
  For further information visit https://errors.pydantic.dev/2.9/v/string_type

我尝试了不同的 pydantic 类型 - str、None、LTreeType，尝试使用验证器，总线仍然成功不是我的，无法将特定数据库类型转换为字符串。

我怎样才能实现这一目标？

谢谢！

Answer 1

我遇到了同样的问题，我通过使用不同的 pydantic 模型解决了这个问题：

import uuid
from re import sub
from typing import Annotated

from fastapi import FastAPI
from pydantic import (
    BaseModel,
    BeforeValidator,
    ConfigDict,
    Field,
    PlainSerializer,
    ValidationError,
    ValidationInfo,
)
from sqlalchemy import (
    Index,
)
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
)
from sqlalchemy_utils import Ltree, LtreeType


def validate_ltree(v: str | Ltree, _info: ValidationInfo):
    if isinstance(v, Ltree):
        return v
    elif isinstance(v, str):
        return Ltree(v)
    raise ValidationError(f"Invalid type: {v.__class__.__name__}")


def serialize_ltree(value: Ltree) -> str:
    return value.path


LtreeField = Annotated[
    Ltree,
    BeforeValidator(validate_ltree),
    PlainSerializer(serialize_ltree),
]


def slugify(s):
    s = s.lower().strip()
    s = sub(r"[^\w\s-]", "", s)
    s = sub(r"[\s_-]+", "_", s)
    s = sub(r"^-+|-+$", "", s)
    return s


class Base(DeclarativeBase):
    pass


class ItemORM(Base):
    __tablename__ = "item"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
    label: Mapped[str]
    path: Mapped[Ltree] = mapped_column(LtreeType, unique=True)

    def __init__(self, label: str, parent_path: str | Ltree = None, **kwargs):
        super().__init__(**kwargs)
        self.id = uuid.uuid4()
        self.label = label
        if parent_path:
            parent_path = Ltree(parent_path)
        ltree_id = Ltree(slugify(self.label))
        self.path = ltree_id if parent_path is None else parent_path + ltree_id

    __table_args__ = (Index("ix_item_path", path, postgresql_using="gist"),)


class ItemBaseDTO(BaseModel):
    pass


class ItemCreateDTO(ItemBaseDTO):
    label: str
    parent_path: str | None = Field(default=None)


class ItemUpdateDTO(ItemBaseDTO):
    label: str | None = Field(default=None)
    parent_path: str | None = Field(default=None)


class ItemPublicDTO(ItemBaseDTO):
    model_config = ConfigDict(from_attributes=False, arbitrary_types_allowed=True)

    id: uuid.UUID
    label: str
    path: LtreeField


app = FastAPI()


@app.get("/", response_model=ItemPublicDTO)
def read_item():
    item = ItemORM(label="Demo")
    print("item\t\t", f"{item!r}")
    print("item.path\t", f"{item.path!r}")
    return item


@app.post("/", response_model=ItemPublicDTO)
def create_item(*, item_in: ItemCreateDTO):
    item = ItemORM(**item_in.model_dump())
    print("item\t\t", f"{item!r}")
    print("item.path\t", f"{item.path!r}")
    return item