如何使工单查询返回父/子工单ID列表而非关联字典列表？

问题描述 #

我开发了一个工单管理应用，工单之间支持交叉引用，用专用表存储关联关系。简化后的SQLAlchemy表结构如下：

class Ticket(Base):
    """Ticket."""
    __tablename__ = "tickets"
    id = Column(Integer, primary_key=True, autoincrement="auto")
    title = Column(String(50), nullable=False)
    children_tickets = relationship(
        "TicketReferences",
        primaryjoin="Ticket.id == TicketReferences.parent_id",
        cascade="all, delete-orphan"
    )
    parents_tickets = relationship(
        "TicketReferences",
        primaryjoin="Ticket.id == TicketReferences.child_id",
        cascade="all, delete-orphan"
    )


class TicketReferences(Base):
    """Tickets can be cross-referenced"""
    __tablename__ = "tickets_references"
    id = Column(Integer, primary_key=True, autoincrement="auto")
    parent_id = Column(Integer, ForeignKey("tickets.id"), nullable=False)
    child_id = Column(Integer, ForeignKey("tickets.id"), nullable=False)

对应的Pydantic Schema：

class TicketsReferences(BaseModel):
    parent_id: int
    child_id: int

    class Config:
        orm_mode = True

class TicketCreate(TicketBase):
    title: str
    parents_tickets: Optional[List[int]] = []
    children_tickets: Optional[List[int]] = []

class Ticket(TicketBase):
    id: int
    title: str
    parents_tickets: List[TicketsReferences]
    children_tickets: List[TicketsReferences]
    class Config:
        orm_mode = True

创建工单的请求体示例：

# 创建第一个工单（ID会是1）
{
  "title": "Foo",
  "parents_tickets": [],
  "children_tickets": []
}

# 创建第二个工单，关联工单1为父工单
{
  "title": "Bar",
  "parents_tickets": [1],
  "children_tickets": []
}

现在查询工单时，返回的关联字段结构过于复杂：

# 查询工单#2的结果
{
  "id": 2,
  "title": "bar",
  "parents_tickets": [
    {
      "parent_id": 1,
      "child_id": 2
    }
  ],
  "children_tickets": []
}

我期望返回的是ID列表，比如"parents_tickets": [1]，请问怎么实现？

解决方案 #

方法1：修改Pydantic模型，转换关联数据 #

如果不想改动SQLAlchemy的关联结构，可以在Pydantic模型中添加转换逻辑，把TicketReferences对象列表转换成纯ID列表。

适配Pydantic v2版本

使用field_validator处理数据转换，同时保留原字段名：

from pydantic import BaseModel, field_validator
from typing import List, Optional

class Ticket(TicketBase):
    id: int
    title: str
    parents_tickets: List[int]
    children_tickets: List[int]

    @field_validator('parents_tickets', mode='before')
    def parse_parents(cls, v):
        # 从SQLAlchemy的关联对象中提取parent_id
        if isinstance(v, list) and all(isinstance(item, TicketReferences) for item in v):
            return [item.parent_id for item in v]
        return v
    
    @field_validator('children_tickets', mode='before')
    def parse_children(cls, v):
        if isinstance(v, list) and all(isinstance(item, TicketReferences) for item in v):
            return [item.child_id for item in v]
        return v

    class Config:
        orm_mode = True

适配Pydantic v1版本

使用@validator装饰器实现相同逻辑：

from pydantic import BaseModel, validator
from typing import List, Optional

class Ticket(TicketBase):
    id: int
    title: str
    parents_tickets: List[int]
    children_tickets: List[int]

    @validator('parents_tickets', pre=True)
    def parse_parents(cls, v):
        if isinstance(v, list) and all(isinstance(item, TicketReferences) for item in v):
            return [item.parent_id for item in v]
        return v
    
    @validator('children_tickets', pre=True)
    def parse_children(cls, v):
        if isinstance(v, list) and all(isinstance(item, TicketReferences) for item in v):
            return [item.child_id for item in v]
        return v

    class Config:
        orm_mode = True

方法2：将SQLAlchemy关联改为多对多自关联 #

更优雅的方式是重构关联关系，使用SQLAlchemy的多对多自关联，让框架直接维护关联表，简化数据转换逻辑。

首先修改SQLAlchemy模型：

class Ticket(Base):
    """Ticket."""
    __tablename__ = "tickets"
    id = Column(Integer, primary_key=True, autoincrement="auto")
    title = Column(String(50), nullable=False)
    # 多对多自关联：父工单关联子工单
    children_tickets = relationship(
        "Ticket",
        secondary="tickets_references",
        primaryjoin="Ticket.id == TicketReferences.parent_id",
        secondaryjoin="Ticket.id == TicketReferences.child_id",
        back_populates="parents_tickets",
        cascade="all, delete-orphan"
    )
    # 反向关联：子工单关联父工单
    parents_tickets = relationship(
        "Ticket",
        secondary="tickets_references",
        primaryjoin="Ticket.id == TicketReferences.child_id",
        secondaryjoin="Ticket.id == TicketReferences.parent_id",
        back_populates="children_tickets",
        cascade="all, delete-orphan"
    )

class TicketReferences(Base):
    """Tickets can be cross-referenced"""
    __tablename__ = "tickets_references"
    # 用复合主键避免重复关联
    parent_id = Column(Integer, ForeignKey("tickets.id"), nullable=False, primary_key=True)
    child_id = Column(Integer, ForeignKey("tickets.id"), nullable=False, primary_key=True)

然后修改Pydantic模型，通过from_orm方法直接提取ID：

class TicketCreate(TicketBase):
    title: str
    parents_tickets: Optional[List[int]] = []
    children_tickets: Optional[List[int]] = []

class Ticket(TicketBase):
    id: int
    title: str
    parents_tickets: List[int]
    children_tickets: List[int]

    class Config:
        orm_mode = True
        
        @classmethod
        def from_orm(cls, obj):
            return cls(
                id=obj.id,
                title=obj.title,
                parents_tickets=[ticket.id for ticket in obj.parents_tickets],
                children_tickets=[ticket.id for ticket in obj.children_tickets]
            )

这样查询工单时就能直接返回纯ID列表，同时创建工单时只需将传入的ID列表转换为对应的Ticket对象添加到关联中即可。

内容的提问来源于stack exchange，提问作者Shan-x