dadao
30caaaf1d9
- item.kind='ticket' + ticket_ext 1:1，迁移 0003 同时重写 item 状态约束为任务∪工单并集 - 工单状态机与动作权限：受理/开工/提交限办理人，验收限需求人（办理人不能自验） - 升级时原任务转 blocked，验收通过后自动回流 todo，事件流两侧都留痕 - item_link（父子/相关）与 notification（只发当事人），三张表均开 RLS - 抽出 serializers.py 复用任务序列化，修复工单响应缺 tags 导致前端白屏 - 评论与 PATCH 放开给工单；conftest 清库前终止残留事务，避免 TRUNCATE 死等锁 - tests/test_tickets.py 11 个用例，全量 57 passed
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""统一的 item / tag 序列化。
|
|
|
|
为什么要单独一份：工单接口最初自己拼了一份 dict，忘了 `tags` 字段，
|
|
前端列表把 `item.tags` 当数组用，一升级成工单就整页白屏。
|
|
任何返回 item 的接口都必须用这里的 `item_out`，并用 `tags_of` 把标签一起取回来。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from .models_task import Item, ItemTag, Tag
|
|
|
|
|
|
def tag_out(tag: Tag) -> dict[str, Any]:
|
|
return {"id": str(tag.id), "name": tag.name, "color": tag.color}
|
|
|
|
|
|
def item_out(item: Item, tags: list[Tag] | None = None) -> dict[str, Any]:
|
|
return {
|
|
"id": str(item.id),
|
|
"kind": item.kind,
|
|
"title": item.title,
|
|
"body_md": item.body_md,
|
|
"status": item.status,
|
|
"priority": item.priority,
|
|
"creator_id": str(item.creator_id),
|
|
"assignee_id": str(item.assignee_id) if item.assignee_id else None,
|
|
"due_at": item.due_at,
|
|
"completed_at": item.completed_at,
|
|
"sort_order": item.sort_order,
|
|
"deleted_at": item.deleted_at,
|
|
"tags": [tag_out(t) for t in (tags or [])],
|
|
"created_at": item.created_at,
|
|
"updated_at": item.updated_at,
|
|
"version": item.version,
|
|
}
|
|
|
|
|
|
async def tags_of(
|
|
session: AsyncSession, ws_id: uuid.UUID, item_ids: list[uuid.UUID]
|
|
) -> dict[uuid.UUID, list[Tag]]:
|
|
"""一次把这些事项的标签查回来（别在循环里逐条查）。"""
|
|
if not item_ids:
|
|
return {}
|
|
rows = (
|
|
await session.execute(
|
|
select(ItemTag.item_id, Tag)
|
|
.join(Tag, Tag.id == ItemTag.tag_id)
|
|
.where(ItemTag.workspace_id == ws_id, ItemTag.item_id.in_(item_ids))
|
|
.order_by(Tag.name)
|
|
)
|
|
).all()
|
|
grouped: dict[uuid.UUID, list[Tag]] = {}
|
|
for item_id, tag in rows:
|
|
grouped.setdefault(item_id, []).append(tag)
|
|
return grouped
|