Files
DadaoTask/backend/app/api/items.py
T
dadao c4868f04cf feat(task): 批量改状态与看板数据接口
- POST /items/batch-status:逐条成败,不做全回滚;拖 10 条里 1 条冲突不该让另 9 条白干
- GET /items/board:四列一次取全,收件箱不进看板
- 修:/items/board 被 /{item_id} 抢先匹配导致 422
- 新增向量库配置项(M3 用,独立实例,不动 fnsvr-01)
- 3 个新测试,全量 46 passed
2026-10-28 04:05:35 +08:00

746 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""任务(item)接口。
写操作的并发控制统一走**乐观锁**:请求必须带 `version`,落库时用
`UPDATE ... WHERE id = ? AND version = ?` 并检查 rowcount。
不用"先查再改"——两个请求同时读到 version=3 时,两边都会认为自己是对的,
后写的静默覆盖先写的,而且日志里看不出异常。
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Annotated, Any
from fastapi import APIRouter, Query
from pydantic import BaseModel, Field
from sqlalchemy import Select, func, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from ..config import get_settings
from ..deps import MemberDep, SessionDep
from ..domain.task import (
ALL_STATUSES,
apply_status,
due_window,
ensure_transition,
view_statuses,
)
from ..errors import AppError, ErrorCode
from ..models_task import Activity, Comment, Item, ItemTag, Tag
from ..schemas import Pagination, encode_cursor, ok
router = APIRouter(prefix="/workspaces/{ws_id}/items", tags=["任务"])
tag_router = APIRouter(prefix="/workspaces/{ws_id}/tags", tags=["标签"])
MAX_TITLE = 200
MAX_BODY = 100_000
# ── 入参 ────────────────────────────────────────────────────────────────
class CreateItemIn(BaseModel):
title: str = Field(min_length=1, max_length=MAX_TITLE)
body_md: str = Field(default="", max_length=MAX_BODY)
status: str = Field(default="inbox")
priority: int = Field(default=0, ge=0, le=3)
assignee_id: uuid.UUID | None = None
due_at: datetime | None = None
tag_names: list[str] = Field(default_factory=list, max_length=10)
class UpdateItemIn(BaseModel):
"""只传要改的字段。注意:`None` 表示"不动",想把 due_at 清空请传 `clear_due_at`。"""
version: int = Field(ge=1)
title: str | None = Field(default=None, min_length=1, max_length=MAX_TITLE)
body_md: str | None = Field(default=None, max_length=MAX_BODY)
priority: int | None = Field(default=None, ge=0, le=3)
assignee_id: uuid.UUID | None = None
clear_assignee: bool = False
due_at: datetime | None = None
clear_due_at: bool = False
tag_names: list[str] | None = Field(default=None, max_length=10)
class StatusIn(BaseModel):
version: int = Field(ge=1)
status: str
class CommentIn(BaseModel):
body_md: str = Field(min_length=1, max_length=20_000)
class TagIn(BaseModel):
name: str = Field(min_length=1, max_length=32)
color: str = Field(default="", max_length=16)
# ── 序列化 ──────────────────────────────────────────────────────────────
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 _load_item(session: AsyncSession, ws_id: uuid.UUID, item_id: uuid.UUID) -> Item:
item = await session.get(Item, item_id)
if item is None or item.workspace_id != ws_id or item.kind != "task":
# 跨租户 / 不存在 / 不是任务,对外一律"不存在",不泄漏别人有没有这条数据
raise AppError(ErrorCode.ITEM_NOT_FOUND, "任务不存在")
return item
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
def _normalize_tag_names(names: list[str]) -> list[str]:
result: list[str] = []
seen: set[str] = set()
for raw in names:
name = raw.strip()
if not name or name.lower() in seen:
continue
seen.add(name.lower())
result.append(name)
return result
async def _set_tags(session: AsyncSession, ws_id: uuid.UUID, item: Item, names: list[str]) -> None:
"""整组替换:客户端传什么就是什么,不做增量(增量语义在弱网下容易和用户预期错位)。"""
names = _normalize_tag_names(names)
existing = {
tag.name.lower(): tag
for tag in (await session.execute(select(Tag).where(Tag.workspace_id == ws_id))).scalars()
}
keep: list[Tag] = []
for name in names:
tag = existing.get(name.lower())
if tag is None:
tag = Tag(workspace_id=ws_id, name=name)
session.add(tag)
await session.flush()
existing[name.lower()] = tag
keep.append(tag)
await session.execute(
ItemTag.__table__.delete().where(ItemTag.workspace_id == ws_id, ItemTag.item_id == item.id)
)
for tag in keep:
session.add(ItemTag(workspace_id=ws_id, item_id=item.id, tag_id=tag.id))
await session.flush()
async def _log(
session: AsyncSession, item: Item, actor_id: uuid.UUID, action: str, payload: dict | None = None
) -> None:
session.add(
Activity(
workspace_id=item.workspace_id,
item_id=item.id,
actor_id=actor_id,
action=action,
payload=payload or {},
)
)
async def _apply_update(
session: AsyncSession,
item: Item,
*,
version: int,
values: dict[str, Any],
actor_id: uuid.UUID,
action: str,
payload: dict | None = None,
) -> Item:
"""乐观锁写库:rowcount=0 说明别人先改了,别默默覆盖。"""
values = {**values, "version": version + 1, "updated_at": datetime.now(timezone.utc)}
result = await session.execute(
update(Item)
.where(Item.id == item.id, Item.workspace_id == item.workspace_id, Item.version == version)
.values(**values)
)
if result.rowcount == 0:
await session.refresh(item)
raise AppError(
ErrorCode.VERSION_CONFLICT,
"这条任务已被别人修改,请刷新后重试",
details={"current_version": item.version},
)
await session.refresh(item)
await _log(session, item, actor_id, action, payload)
return item
# ── 列表 ────────────────────────────────────────────────────────────────
def _apply_filters(
stmt: Select,
*,
ws_id: uuid.UUID,
me: uuid.UUID,
view: str | None,
status: list[str] | None,
assignee: str | None,
q: str | None,
tag: str | None,
include_deleted: bool,
now: datetime,
tz: str,
) -> Select:
stmt = stmt.where(Item.workspace_id == ws_id, Item.kind == "task")
if view == "deleted":
stmt = stmt.where(Item.deleted_at.is_not(None))
elif not include_deleted:
stmt = stmt.where(Item.deleted_at.is_(None))
if status:
invalid = [s for s in status if s not in ALL_STATUSES]
if invalid:
raise AppError(ErrorCode.BAD_REQUEST, f"未知状态:{','.join(invalid)}")
stmt = stmt.where(Item.status.in_(status))
elif view and view != "deleted":
stmt = stmt.where(Item.status.in_(view_statuses(view)))
due_from, due_to = due_window(view, now=now, tz=tz)
if due_from is not None:
stmt = stmt.where(Item.due_at.is_not(None), Item.due_at >= due_from, Item.due_at < due_to)
if assignee:
if assignee == "me":
stmt = stmt.where(Item.assignee_id == me)
else:
try:
stmt = stmt.where(Item.assignee_id == uuid.UUID(assignee))
except ValueError as exc:
raise AppError(ErrorCode.BAD_REQUEST, "assignee 需要是 me 或用户 id") from exc
if q and q.strip():
needle = f"%{q.strip()}%"
stmt = stmt.where(or_(Item.title.ilike(needle), Item.body_md.ilike(needle)))
if tag and tag.strip():
stmt = stmt.where(
Item.id.in_(
select(ItemTag.item_id)
.join(Tag, Tag.id == ItemTag.tag_id)
.where(ItemTag.workspace_id == ws_id, func.lower(Tag.name) == tag.strip().lower())
)
)
return stmt
@router.get("", summary="任务列表(视图 / 状态 / 负责人 / 关键字 / 标签)")
async def list_items(
ctx: MemberDep,
session: SessionDep,
view: Annotated[
str | None, Query(description="inbox|active|today|week|done|canceled|all|deleted")
] = None,
status: Annotated[list[str] | None, Query(description="精确状态,可重复传")] = None,
assignee: Annotated[str | None, Query(description="me 或用户 uuid")] = None,
q: Annotated[str | None, Query(max_length=100, description="标题/正文关键字")] = None,
tag: Annotated[str | None, Query(max_length=32)] = None,
sort: Annotated[str, Query(description="updated|due|created")] = "updated",
include_deleted: bool = False,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[str | None, Query()] = None,
) -> dict:
if sort not in ("updated", "due", "created"):
raise AppError(ErrorCode.BAD_REQUEST, "sort 只能是 updated / due / created")
settings = get_settings()
now = datetime.now(timezone.utc)
page = Pagination(limit=limit, cursor=cursor)
limit, parsed_cursor = page.parsed()
base = _apply_filters(
select(Item),
ws_id=ctx.ws_id,
me=ctx.user.id,
view=view,
status=status,
assignee=assignee,
q=q,
tag=tag,
include_deleted=include_deleted,
now=now,
tz=settings.default_timezone,
)
if sort == "due":
# 截止时间排序涉及 NULLS LAST,keyset 分页会把 NULL 行切碎。
# 逾期/今日清单本来就是短列表,这里不分页(上限 100 条),语义更清楚。
rows = (
(
await session.execute(
base.order_by(Item.due_at.asc().nulls_last(), Item.sort_order, Item.id).limit(
limit
)
)
)
.scalars()
.all()
)
tags = await _tags_of(session, ctx.ws_id, [r.id for r in rows])
return ok(
{
"items": [_item_out(r, tags.get(r.id)) for r in rows],
"next_cursor": None,
"has_more": False,
}
)
if sort == "created":
order = (Item.created_at.desc(), Item.id.desc())
sort_col, sort_attr = Item.created_at, "created_at"
else:
order = (Item.updated_at.desc(), Item.id.desc())
sort_col, sort_attr = Item.updated_at, "updated_at"
stmt = base.order_by(*order).limit(limit + 1)
if parsed_cursor is not None:
stmt = stmt.where(
or_(
sort_col < parsed_cursor.created_at,
(sort_col == parsed_cursor.created_at) & (Item.id < parsed_cursor.id),
)
)
rows = (await session.execute(stmt)).scalars().all()
has_more = len(rows) > limit
rows = list(rows[:limit])
next_cursor = (
encode_cursor(getattr(rows[-1], sort_attr), rows[-1].id) if has_more and rows else None
)
tags = await _tags_of(session, ctx.ws_id, [r.id for r in rows])
return ok(
{
"items": [_item_out(r, tags.get(r.id)) for r in rows],
"next_cursor": next_cursor,
"has_more": has_more,
}
)
@router.get("/board", summary="看板:按状态分列的最近任务")
async def board(
ctx: MemberDep,
session: SessionDep,
per_column: Annotated[int, Query(ge=1, le=100)] = 30,
) -> dict:
"""一屏拖动需要的数据,一次拿全。
每列按"截止时间早的在前、没有截止时间的靠后"排,其次按更新时间。
只取 `per_column` 条:看板是"手上正在做的",不是全量列表。
"""
columns: dict[str, list[dict[str, Any]]] = {}
statuses = ("todo", "doing", "blocked", "done")
rows = (
(
await session.execute(
select(Item)
.where(
Item.workspace_id == ctx.ws_id,
Item.status.in_(statuses),
Item.deleted_at.is_(None),
)
.order_by(
Item.due_at.asc().nulls_last(),
Item.updated_at.desc(),
Item.id.desc(),
)
.limit(per_column * len(statuses))
)
)
.scalars()
.all()
)
by_id = {row.id: row for row in rows}
tag_map = await _tags_of(session, ctx.ws_id, list(by_id))
for status in statuses:
columns[status] = [
_item_out(row, tag_map.get(row.id)) for row in rows if row.status == status
][:per_column]
return ok(
{
"columns": columns,
"order": list(statuses),
"labels": {
"todo": "待办",
"doing": "进行中",
"blocked": "被阻塞",
"done": "已完成",
},
}
)
@router.get("/{item_id}", summary="任务详情(含最近 50 条事件)")
async def get_item(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
item = await _load_item(session, ctx.ws_id, item_id)
tags = await _tags_of(session, ctx.ws_id, [item.id])
activities = (
(
await session.execute(
select(Activity)
.where(Activity.workspace_id == ctx.ws_id, Activity.item_id == item.id)
.order_by(Activity.created_at.desc(), Activity.id.desc())
.limit(50)
)
)
.scalars()
.all()
)
data = _item_out(item, tags.get(item.id))
data["activities"] = [
{
"id": str(a.id),
"action": a.action,
"actor_id": str(a.actor_id) if a.actor_id else None,
"payload": a.payload,
"created_at": a.created_at,
}
for a in activities
]
return ok(data)
# ── 写操作 ──────────────────────────────────────────────────────────────
@router.post("", summary="新建任务")
async def create_item(body: CreateItemIn, ctx: MemberDep, session: SessionDep) -> dict:
if body.status not in ALL_STATUSES:
raise AppError(ErrorCode.BAD_REQUEST, f"未知状态:{body.status}")
fields: dict[str, Any] = {"status": body.status, "completed_at": None}
apply_status(fields, body.status, now=datetime.now(timezone.utc))
item = Item(
id=uuid.uuid4(),
workspace_id=ctx.ws_id,
kind="task",
title=body.title.strip(),
body_md=body.body_md,
priority=body.priority,
creator_id=ctx.user.id,
assignee_id=body.assignee_id,
due_at=body.due_at,
status=fields["status"],
completed_at=fields["completed_at"],
)
session.add(item)
await session.flush()
if body.tag_names:
await _set_tags(session, ctx.ws_id, item, body.tag_names)
await _log(session, item, ctx.user.id, "created", {"status": item.status})
tags = await _tags_of(session, ctx.ws_id, [item.id])
return ok(_item_out(item, tags.get(item.id)), message="已创建")
class BatchStatusItemIn(BaseModel):
id: uuid.UUID
version: int
class BatchStatusIn(BaseModel):
status: str
items: list[BatchStatusItemIn] = Field(min_length=1, max_length=50)
@router.post("/batch-status", summary="批量改状态(逐条返回成败)")
async def batch_status(body: BatchStatusIn, ctx: MemberDep, session: SessionDep) -> dict:
"""看板多选拖拽用。
不做"全成功或全回滚":拖 10 条里有 1 条被同事改过,其余 9 条应当照样生效,
失败的那条单独告诉前端原因。全回滚会让用户白干一次,而且更难解释。
"""
if body.status not in ALL_STATUSES:
raise AppError(ErrorCode.BAD_REQUEST, f"未知状态:{body.status}")
now = datetime.now(timezone.utc)
updated: list[dict[str, Any]] = []
failed: list[dict[str, Any]] = []
for entry in body.items:
try:
item = await _load_item(session, ctx.ws_id, entry.id)
except AppError as exc:
failed.append({"id": str(entry.id), "code": int(exc.code), "message": exc.message})
continue
if item.status == body.status:
updated.append({"id": str(item.id), "version": item.version, "unchanged": True})
continue
try:
ensure_transition(item.status, body.status)
except AppError as exc:
failed.append(
{
"id": str(entry.id),
"code": int(exc.code),
"message": exc.message,
"allowed": (exc.details or {}).get("allowed"),
}
)
continue
previous = item.status
fields: dict[str, Any] = {"status": body.status, "completed_at": None}
apply_status(fields, body.status, now=now)
try:
item = await _apply_update(
session,
item,
version=entry.version,
values={"status": fields["status"], "completed_at": fields["completed_at"]},
actor_id=ctx.user.id,
action="status_changed",
payload={"from": previous, "to": body.status, "batch": True},
)
except AppError as exc:
failed.append(
{
"id": str(entry.id),
"code": int(exc.code),
"message": exc.message,
"current_version": (exc.details or {}).get("current_version"),
}
)
continue
updated.append({"id": str(item.id), "version": item.version, "unchanged": False})
return ok(
{"updated": updated, "failed": failed},
message=f"已更新 {len(updated)} 条,{len(failed)} 条未生效",
)
@router.patch("/{item_id}", summary="修改任务字段(乐观锁)")
async def update_item(
item_id: uuid.UUID, body: UpdateItemIn, ctx: MemberDep, session: SessionDep
) -> dict:
item = await _load_item(session, ctx.ws_id, item_id)
values: dict[str, Any] = {}
changes: dict[str, Any] = {}
if body.title is not None and body.title.strip() != item.title:
values["title"] = body.title.strip()
changes["title"] = values["title"]
if body.body_md is not None and body.body_md != item.body_md:
values["body_md"] = body.body_md
changes["body_md"] = "changed"
if body.priority is not None and body.priority != item.priority:
values["priority"] = body.priority
changes["priority"] = body.priority
if body.clear_assignee:
values["assignee_id"] = None
changes["assignee_id"] = None
elif body.assignee_id is not None and body.assignee_id != item.assignee_id:
values["assignee_id"] = body.assignee_id
changes["assignee_id"] = str(body.assignee_id)
if body.clear_due_at:
values["due_at"] = None
changes["due_at"] = None
elif body.due_at is not None and body.due_at != item.due_at:
values["due_at"] = body.due_at
changes["due_at"] = body.due_at.isoformat()
item = await _apply_update(
session,
item,
version=body.version,
values=values,
actor_id=ctx.user.id,
action="updated",
payload={"changes": changes} if changes else None,
)
if body.tag_names is not None:
await _set_tags(session, ctx.ws_id, item, body.tag_names)
await _log(
session,
item,
ctx.user.id,
"tags_changed",
{"tags": _normalize_tag_names(body.tag_names)},
)
item.version += 1
tags = await _tags_of(session, ctx.ws_id, [item.id])
return ok(_item_out(item, tags.get(item.id)), message="已保存")
@router.post("/{item_id}/status", summary="改状态(状态机 + 乐观锁)")
async def change_status(
item_id: uuid.UUID, body: StatusIn, ctx: MemberDep, session: SessionDep
) -> dict:
item = await _load_item(session, ctx.ws_id, item_id)
if body.status not in ALL_STATUSES:
raise AppError(ErrorCode.BAD_REQUEST, f"未知状态:{body.status}")
if body.status == item.status:
tags = await _tags_of(session, ctx.ws_id, [item.id])
return ok(_item_out(item, tags.get(item.id)), message="状态未变化")
ensure_transition(item.status, body.status)
previous = item.status
fields: dict[str, Any] = {"status": body.status, "completed_at": None}
apply_status(fields, body.status, now=datetime.now(timezone.utc))
item = await _apply_update(
session,
item,
version=body.version,
values={"status": fields["status"], "completed_at": fields["completed_at"]},
actor_id=ctx.user.id,
action="status_changed",
payload={"from": previous, "to": body.status},
)
tags = await _tags_of(session, ctx.ws_id, [item.id])
return ok(_item_out(item, tags.get(item.id)), message="状态已更新")
@router.delete("/{item_id}", summary="删除任务(软删;带 version 才防并发)")
async def delete_item(
item_id: uuid.UUID,
ctx: MemberDep,
session: SessionDep,
version: Annotated[int, Query(ge=1)],
) -> dict:
item = await _load_item(session, ctx.ws_id, item_id)
item = await _apply_update(
session,
item,
version=version,
values={"deleted_at": datetime.now(timezone.utc)},
actor_id=ctx.user.id,
action="deleted",
)
return ok({"id": str(item.id), "deleted_at": item.deleted_at}, message="已删除")
# ── 评论 ────────────────────────────────────────────────────────────────
@router.post("/{item_id}/comments", summary="加评论")
async def add_comment(
item_id: uuid.UUID, body: CommentIn, ctx: MemberDep, session: SessionDep
) -> dict:
item = await _load_item(session, ctx.ws_id, item_id)
comment = Comment(
workspace_id=ctx.ws_id,
item_id=item.id,
author_id=ctx.user.id,
body_md=body.body_md,
)
session.add(comment)
await session.flush()
await _log(session, item, ctx.user.id, "commented", {"comment_id": str(comment.id)})
return ok(
{
"id": str(comment.id),
"item_id": str(item.id),
"author_id": str(comment.author_id),
"body_md": comment.body_md,
"created_at": comment.created_at,
},
message="已评论",
)
@router.get("/{item_id}/comments", summary="评论列表")
async def list_comments(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
item = await _load_item(session, ctx.ws_id, item_id)
rows = (
(
await session.execute(
select(Comment)
.where(
Comment.workspace_id == ctx.ws_id,
Comment.item_id == item.id,
Comment.deleted_at.is_(None),
)
.order_by(Comment.created_at.asc(), Comment.id.asc())
.limit(200)
)
)
.scalars()
.all()
)
return ok(
{
"items": [
{
"id": str(c.id),
"author_id": str(c.author_id),
"body_md": c.body_md,
"created_at": c.created_at,
}
for c in rows
]
}
)
# ── 标签 ────────────────────────────────────────────────────────────────
@tag_router.get("", summary="标签列表")
async def list_tags(ctx: MemberDep, session: SessionDep) -> dict:
rows = (
(await session.execute(select(Tag).where(Tag.workspace_id == ctx.ws_id).order_by(Tag.name)))
.scalars()
.all()
)
return ok({"items": [_tag_out(t) for t in rows]})
@tag_router.post("", summary="建标签(同名返回已有)")
async def create_tag(body: TagIn, ctx: MemberDep, session: SessionDep) -> dict:
name = body.name.strip()
exists = (
await session.execute(
select(Tag).where(Tag.workspace_id == ctx.ws_id, func.lower(Tag.name) == name.lower())
)
).scalar_one_or_none()
if exists is not None:
return ok(_tag_out(exists), message="已存在")
tag = Tag(workspace_id=ctx.ws_id, name=name, color=body.color)
session.add(tag)
await session.flush()
return ok(_tag_out(tag), message="已创建")