feat(task): M1 任务域后端与 Web 列表(状态机/乐观锁/视图/搜索)
CI / backend (push) Canceled after 0s

- item/tag/item_tag/comment/activity 五表迁移,RLS 全覆盖,pg_trgm 子串索引
- 状态机 domain/task.py:非法流转回 5002 并给 allowed,done 写 completed_at
- /workspaces/{ws}/items 列表/详情/新建/改/状态/软删/评论/标签
- 乐观锁 UPDATE...WHERE version,冲突 4009 + current_version
- Web 任务页:视图 tab、快速新建、勾选完成、开工/阻塞、详情事件流、游标分页
- 测试 43 全绿,ruff 干净,Web 构建通过
- 文档:2.Design/API.md 新增,ROADMAP 按已验证口径重写
This commit is contained in:
Steven
2026-10-28 03:48:48 +08:00
parent d390b7b58a
commit e25a1a4e31
19 changed files with 1959 additions and 31 deletions
+202
View File
@@ -0,0 +1,202 @@
"""M1 任务域:item / tag / item_tag / comment / activity + 行级安全。
RLS 策略与 0001 同一套模式:`workspace_id = app_ws_id()`,读写都校验。
`item` 额外允许"我创建或分配给我"——不,**不做这个特例**:跨租户看别人的任务没有业务含义,
越权面越小越好,列表查询本来就带 workspace_id 过滤。
注意:`pg_trgm` 是 trusted extension(PG13+),库 owner 就能装,所以迁移不需要 superuser。
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "0002"
down_revision: str | None = "0001"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
APP_ROLE = "dadao_app"
TENANT_TABLES_0002: tuple[str, ...] = ("item", "tag", "item_tag", "comment", "activity")
_TIMESTAMP_COLUMNS = (
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.Column("version", sa.Integer, nullable=False, server_default="1"),
)
def _ws_fk() -> sa.Column:
return sa.Column(
"workspace_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("workspace.id", ondelete="CASCADE"),
nullable=False,
)
def upgrade() -> None:
# 中文没有分词器(zhparser 未装),M1 的搜索走 ILIKE 子串匹配,
# trigram 索引能让"标题里找关键字"这条最常用的查询走索引而不是全表扫。
op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
op.create_table(
"item",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
_ws_fk(),
sa.Column("kind", sa.String(16), nullable=False, server_default="task"),
sa.Column("title", sa.String(200), nullable=False),
sa.Column("body_md", sa.Text, nullable=False, server_default=""),
sa.Column("status", sa.String(16), nullable=False, server_default="inbox"),
sa.Column("priority", sa.Integer, nullable=False, server_default="0"),
sa.Column(
"creator_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("app_user.id", ondelete="RESTRICT"),
nullable=False,
),
sa.Column(
"assignee_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("app_user.id", ondelete="SET NULL"),
),
sa.Column("due_at", sa.DateTime(timezone=True)),
sa.Column("completed_at", sa.DateTime(timezone=True)),
sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"),
sa.Column("deleted_at", sa.DateTime(timezone=True)),
*_TIMESTAMP_COLUMNS,
sa.CheckConstraint(
"status IN ('inbox','todo','doing','done','blocked','canceled')",
name="ck_item_status",
),
sa.CheckConstraint("kind IN ('task','ticket','knowledge')", name="ck_item_kind"),
sa.CheckConstraint("priority BETWEEN 0 AND 3", name="ck_item_priority"),
)
op.create_index("ix_item_ws_status_due", "item", ["workspace_id", "status", "due_at"])
op.create_index("ix_item_ws_assignee", "item", ["workspace_id", "assignee_id", "status"])
op.create_index("ix_item_ws_updated", "item", ["workspace_id", "updated_at"])
op.create_index("ix_item_ws_sort", "item", ["workspace_id", "status", "sort_order"])
op.create_table(
"tag",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
_ws_fk(),
sa.Column("name", sa.String(32), nullable=False),
sa.Column("color", sa.String(16), nullable=False, server_default=""),
*_TIMESTAMP_COLUMNS,
sa.UniqueConstraint("workspace_id", "name", name="uq_tag_ws_name"),
)
op.create_index("ix_tag_ws", "tag", ["workspace_id", "name"])
op.create_table(
"item_tag",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
_ws_fk(),
sa.Column(
"item_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("item.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"tag_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("tag.id", ondelete="CASCADE"),
nullable=False,
),
*_TIMESTAMP_COLUMNS,
sa.UniqueConstraint("item_id", "tag_id", name="uq_item_tag"),
)
op.create_index("ix_item_tag_ws_tag", "item_tag", ["workspace_id", "tag_id"])
op.create_table(
"comment",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
_ws_fk(),
sa.Column(
"item_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("item.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"author_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("app_user.id", ondelete="RESTRICT"),
nullable=False,
),
sa.Column("body_md", sa.Text, nullable=False, server_default=""),
sa.Column("deleted_at", sa.DateTime(timezone=True)),
*_TIMESTAMP_COLUMNS,
)
op.create_index("ix_comment_ws_item_created", "comment", ["workspace_id", "item_id", "created_at"])
op.create_table(
"activity",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
_ws_fk(),
sa.Column(
"item_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("item.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"actor_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("app_user.id", ondelete="SET NULL"),
),
sa.Column("action", sa.String(32), nullable=False),
sa.Column("payload", postgresql.JSONB),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
)
op.create_index("ix_activity_ws_item_created", "activity", ["workspace_id", "item_id", "created_at"])
# 标题三元组索引:ILIKE '%关键字%' 能用到它
op.execute(
"CREATE INDEX ix_item_title_trgm ON item USING gin (title gin_trgm_ops)"
)
for statement in _POLICY_STATEMENTS:
op.execute(statement)
for statement in _GRANT_STATEMENTS:
op.execute(statement)
def downgrade() -> None:
# 先摘策略:策略里引用了这些表本身,表没了策略语句就悬空
for table in TENANT_TABLES_0002:
op.execute(f"DROP POLICY IF EXISTS {table}_isolation ON {table}")
op.execute("DROP INDEX IF EXISTS ix_item_title_trgm")
for table in (
"activity",
"comment",
"item_tag",
"tag",
"item",
):
op.drop_table(table)
# pg_trgm 是库级共享资源,别在这里删(其它模块可能也在用)
_POLICY_STATEMENTS: tuple[str, ...] = tuple(
statement
for table in TENANT_TABLES_0002
for statement in (
f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY;",
f"""
CREATE POLICY {table}_isolation ON {table}
USING (workspace_id = app_ws_id())
WITH CHECK (workspace_id = app_ws_id());
""",
)
)
_GRANT_STATEMENTS: tuple[str, ...] = (
f"GRANT SELECT, INSERT, UPDATE, DELETE ON {', '.join(TENANT_TABLES_0002)} TO {APP_ROLE};",
)
+612
View File
@@ -0,0 +1,612 @@
"""任务(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("/{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="已创建")
@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="已创建")
+5
View File
@@ -42,6 +42,11 @@ class Settings(BaseSettings):
# ── Redis ─────────────────────────────────────────────
redis_url: str = "redis://127.0.0.1:6380/0"
# ── 业务 ──────────────────────────────────────────────
# "今天/本周"这类视图按用户时区切日界,不能按 UTC —— 中国用户 23:30 建的任务
# 在 UTC 下已经是"明天"了。M7 多地域时改成按工作区配置。
default_timezone: str = "Asia/Shanghai"
# ── 认证 ──────────────────────────────────────────────
jwt_secret: str = DEV_JWT_SECRET
jwt_algorithm: str = "HS256"
+1
View File
@@ -0,0 +1 @@
"""任务域业务规则(与框架无关,方便单测)。"""
+100
View File
@@ -0,0 +1,100 @@
"""任务状态机与"视图"口径。
两条原则:
1. **状态机是契约**:客户端据此置灰按钮,服务端必须再校验一次。
任何"直接改 status 列"的代码路径都是 bug。
2. **视图是查询口径,不是状态**:`today` / `week` 是筛选条件,不往 item 上落字段,
否则每天都要批量刷库,还会因为时区边界出错。
"""
from __future__ import annotations
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from ..errors import AppError, ErrorCode
# 允许的流转。没列出的就是禁止的(显式白名单,默认拒绝)。
TRANSITIONS: dict[str, frozenset[str]] = {
# 待整理:进来先放着,决定做或不做
"inbox": frozenset({"todo", "canceled"}),
# 待办:可以开工,也可以直接标记完成(写文档类的事不需要"进行中")
"todo": frozenset({"inbox", "doing", "done", "canceled"}),
# 进行中
"doing": frozenset({"todo", "done", "blocked", "canceled"}),
# 被阻塞(M2 起:任务升级成工单会自动进这里)
"blocked": frozenset({"doing", "canceled"}),
# 已完成:只能重开
"done": frozenset({"todo"}),
# 已取消:只能重开
"canceled": frozenset({"todo"}),
}
# 表示"这件事还没完"的状态
OPEN_STATUSES: tuple[str, ...] = ("inbox", "todo", "doing", "blocked")
ACTIVE_STATUSES: tuple[str, ...] = ("todo", "doing", "blocked")
ALL_STATUSES: tuple[str, ...] = tuple(TRANSITIONS)
# 视图 → 状态集合。`today` / `week` 还要叠加截止时间窗口(见 due_window)
VIEW_STATUSES: dict[str, tuple[str, ...]] = {
"inbox": ("inbox",),
"active": ACTIVE_STATUSES,
"today": ACTIVE_STATUSES,
"week": ACTIVE_STATUSES,
"done": ("done",),
"canceled": ("canceled",),
"all": ALL_STATUSES,
}
DEFAULT_VIEW = "active"
def can_transition(current: str, target: str) -> bool:
return target in TRANSITIONS.get(current, frozenset())
def ensure_transition(current: str, target: str) -> None:
"""非法流转直接拒绝,并把"当前能去哪"一起返回,客户端不用自己猜。"""
if current == target:
return
if not can_transition(current, target):
raise AppError(
ErrorCode.ITEM_STATUS_INVALID,
f"状态不能从 {current} 直接变成 {target}",
details={
"from": current,
"to": target,
"allowed": sorted(TRANSITIONS.get(current, ())),
},
)
def apply_status(item: dict, target: str, *, now: datetime) -> None:
"""就地更新 status/completed_at。`completed_at` 只在 done 时有值。"""
item["status"] = target
item["completed_at"] = now if target == "done" else None
def view_statuses(view: str | None) -> tuple[str, ...]:
return VIEW_STATUSES.get(view or DEFAULT_VIEW, ACTIVE_STATUSES)
def due_window(
view: str | None, *, now: datetime, tz: str
) -> tuple[datetime | None, datetime | None]:
"""返回 (due_from, due_to) 半开区间 `[from, to)`。
`today` = 今天 00:00(含)到明天 00:00(不含),按**用户时区**算,
不是 UTC —— 中国用户 23:30 创建"今天到期"的任务,UTC 下已经是明天了。
"""
if view not in ("today", "week"):
return None, None
zone = ZoneInfo(tz)
local_now = now.astimezone(zone)
start = local_now.replace(hour=0, minute=0, second=0, microsecond=0)
if view == "today":
end = start + timedelta(days=1)
else:
# 本周剩余:到今天所在自然周的周日结束
end = start + timedelta(days=7 - start.weekday())
return start, end
+8
View File
@@ -36,6 +36,11 @@ class ErrorCode(IntEnum):
# 4xxx 并发
VERSION_CONFLICT = 4009
# 5xxx 任务域
ITEM_NOT_FOUND = 5001
ITEM_STATUS_INVALID = 5002
TAG_NOT_FOUND = 5003
# 默认 HTTP 状态:业务码自己带语义,HTTP 只是传输层
_HTTP_STATUS: dict[ErrorCode, int] = {
@@ -59,6 +64,9 @@ _HTTP_STATUS: dict[ErrorCode, int] = {
ErrorCode.ALREADY_A_MEMBER: 409,
ErrorCode.INVITATION_INVALID: 400,
ErrorCode.VERSION_CONFLICT: 409,
ErrorCode.ITEM_NOT_FOUND: 404,
ErrorCode.ITEM_STATUS_INVALID: 409,
ErrorCode.TAG_NOT_FOUND: 404,
}
+3 -1
View File
@@ -12,7 +12,7 @@ from fastapi.responses import JSONResponse
from sqlalchemy import text
from starlette.exceptions import HTTPException as StarletteHTTPException
from .api import auth, workspaces
from .api import auth, items, workspaces
from .config import get_settings
from .db import SessionFactory, current_app_role, engine
from .errors import AppError, ErrorCode, http_status_for
@@ -100,6 +100,8 @@ async def handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
app.include_router(auth.router, prefix=settings.api_prefix)
app.include_router(workspaces.router, prefix=settings.api_prefix)
app.include_router(items.router, prefix=settings.api_prefix)
app.include_router(items.tag_router, prefix=settings.api_prefix)
@app.get("/health", tags=["运维"], summary="存活探针")
+12 -1
View File
@@ -132,4 +132,15 @@ class AuditLog(Base):
# 需要 RLS 的租户表(迁移与测试共用这一份清单,避免漏加)
TENANT_TABLES: tuple[str, ...] = ("workspace", "member", "invitation", "audit_log")
#
# 任务域的表在 models_task.py —— 它 import 本模块的 Base,所以只能在文件末尾反向 import,
# 否则循环导入。
from .models_task import TASK_TENANT_TABLES # noqa: E402
TENANT_TABLES: tuple[str, ...] = (
"workspace",
"member",
"invitation",
"audit_log",
*TASK_TENANT_TABLES,
)
+148
View File
@@ -0,0 +1,148 @@
"""任务域模型(M1):item / tag / item_tag / comment / activity。
约定与 M0 一致:uuid 主键、通用列、`workspace_id NOT NULL`。
两个刻意的取舍:
- **M1 只有一个负责人 `assignee_id`**,不做 `item_assignee` 多对多。M1 是"自己的待办",
多人协作在 M2 工单里才真正需要;提前建多对多表和对应 UI 是过度设计,等真要用再迁移。
- **`kind` 列现在就加上**(默认 `task`)。M2 的工单、M3 的知识都是同一条 item 流,
现在加一列只是默认值,将来加 `ticket_ext` / `knowledge_doc` 时不用改主表。
"""
from __future__ import annotations
import uuid
from datetime import datetime
from sqlalchemy import (
DateTime,
ForeignKey,
Index,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import UUID as PgUUID
from sqlalchemy.orm import Mapped, mapped_column
from .models import Base, Timestamps, new_uuid
ITEM_STATUSES: tuple[str, ...] = ("inbox", "todo", "doing", "done", "blocked", "canceled")
class Item(Base, Timestamps):
"""任务/工单/知识的统一载体。M1 只用 `kind='task'`。"""
__tablename__ = "item"
__table_args__ = (
Index("ix_item_ws_status_due", "workspace_id", "status", "due_at"),
Index("ix_item_ws_assignee", "workspace_id", "assignee_id", "status"),
Index("ix_item_ws_updated", "workspace_id", "updated_at"),
Index("ix_item_ws_sort", "workspace_id", "status", "sort_order"),
)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
kind: Mapped[str] = mapped_column(String(16), nullable=False, default="task")
title: Mapped[str] = mapped_column(String(200), nullable=False)
body_md: Mapped[str] = mapped_column(Text, nullable=False, default="")
status: Mapped[str] = mapped_column(String(16), nullable=False, default="inbox")
priority: Mapped[int] = mapped_column(nullable=False, default=0)
creator_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="RESTRICT"), nullable=False
)
assignee_id: Mapped[uuid.UUID | None] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="SET NULL")
)
due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
sort_order: Mapped[int] = mapped_column(nullable=False, default=0)
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
class Tag(Base, Timestamps):
__tablename__ = "tag"
__table_args__ = (
UniqueConstraint("workspace_id", "name", name="uq_tag_ws_name"),
Index("ix_tag_ws", "workspace_id", "name"),
)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
name: Mapped[str] = mapped_column(String(32), nullable=False)
color: Mapped[str] = mapped_column(String(16), nullable=False, default="")
class ItemTag(Base, Timestamps):
__tablename__ = "item_tag"
__table_args__ = (
UniqueConstraint("item_id", "tag_id", name="uq_item_tag"),
Index("ix_item_tag_ws_tag", "workspace_id", "tag_id"),
)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
item_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), nullable=False
)
tag_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("tag.id", ondelete="CASCADE"), nullable=False
)
class Comment(Base, Timestamps):
__tablename__ = "comment"
__table_args__ = (Index("ix_comment_ws_item_created", "workspace_id", "item_id", "created_at"),)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
item_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), nullable=False
)
author_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="RESTRICT"), nullable=False
)
body_md: Mapped[str] = mapped_column(Text, nullable=False, default="")
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
class Activity(Base):
"""事件流:item 的每一次实质变更都落一条。
这是"这条任务怎么走到今天的"唯一真相源,M4 的 LLM 沉淀、M3 的知识都从这里取料。
只增不改,所以没有 version / updated_at。
"""
__tablename__ = "activity"
__table_args__ = (
Index("ix_activity_ws_item_created", "workspace_id", "item_id", "created_at"),
)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
item_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), nullable=False
)
actor_id: Mapped[uuid.UUID | None] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="SET NULL")
)
action: Mapped[str] = mapped_column(String(32), nullable=False)
payload: Mapped[dict | None] = mapped_column(JSONB)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
TASK_TENANT_TABLES: tuple[str, ...] = ("item", "tag", "item_tag", "comment", "activity")
+3
View File
@@ -7,6 +7,9 @@ pydantic==2.10.4
pydantic-settings==2.7.0
pyjwt==2.10.1
redis==5.2.1
# Windows 没有系统 tzdata,zoneinfo 找不到 "Asia/Shanghai";
# Linux 上装它也无害(会用包里的,而不是系统的)。
tzdata==2026.5
# dev / test only (split into requirements-dev.txt when it grows)
pytest==8.3.4
+272
View File
@@ -0,0 +1,272 @@
"""任务域:状态机、乐观锁、视图、搜索、租户隔离。
这些用例覆盖的是"用户能感觉到的错",不是代码覆盖率:
状态乱跳、两个人同时改、别人的任务突然出现、今天的清单里混进明天的任务。
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
from httpx import AsyncClient
from .conftest import create_workspace, login, unwrap
TZ = ZoneInfo("Asia/Shanghai")
def items_url(ws_id: str) -> str:
return f"/api/v1/workspaces/{ws_id}/items"
async def _new_item(client: AsyncClient, user: dict, ws_id: str, **overrides) -> dict:
payload = {"title": "写周报", **overrides}
return unwrap(await client.post(items_url(ws_id), json=payload, headers=user["headers"]))
async def test_create_task_defaults_to_inbox(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
item = await _new_item(client, user, ws["id"])
assert item["status"] == "inbox"
assert item["kind"] == "task"
assert item["version"] == 1
assert item["creator_id"] == user["user_id"]
inbox = unwrap(
await client.get(items_url(ws["id"]), params={"view": "inbox"}, headers=user["headers"])
)
assert [i["id"] for i in inbox["items"]] == [item["id"]]
async def test_status_machine_happy_path_sets_completed_at(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
item = await _new_item(client, user, ws["id"])
for target in ("todo", "doing", "done"):
item = unwrap(
await client.post(
f"{items_url(ws['id'])}/{item['id']}/status",
json={"status": target, "version": item["version"]},
headers=user["headers"],
)
)
assert item["status"] == target
assert item["completed_at"] is not None
# 重开:completed_at 必须被清掉,否则"完成时间"会一直挂着
item = unwrap(
await client.post(
f"{items_url(ws['id'])}/{item['id']}/status",
json={"status": "todo", "version": item["version"]},
headers=user["headers"],
)
)
assert item["status"] == "todo"
assert item["completed_at"] is None
async def test_illegal_transition_is_rejected(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
item = await _new_item(client, user, ws["id"])
resp = await client.post(
f"{items_url(ws['id'])}/{item['id']}/status",
json={"status": "done", "version": item["version"]}, # inbox 不能直接 done
headers=user["headers"],
)
body = resp.json()
assert body["code"] == 5002
assert body["data"]["from"] == "inbox"
assert "todo" in body["data"]["allowed"]
async def test_stale_version_is_rejected_not_silently_overwritten(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
item = await _new_item(client, user, ws["id"])
first = unwrap(
await client.patch(
f"{items_url(ws['id'])}/{item['id']}",
json={"version": item["version"], "title": "写周报(周四)"},
headers=user["headers"],
)
)
assert first["version"] == item["version"] + 1
resp = await client.patch(
f"{items_url(ws['id'])}/{item['id']}",
json={"version": item["version"], "title": "覆盖别人的修改"},
headers=user["headers"],
)
body = resp.json()
assert resp.status_code == 409
assert body["code"] == 4009
assert body["data"]["current_version"] == first["version"]
async def test_other_tenant_cannot_read_or_list_my_task(client: AsyncClient) -> None:
me = await login(client)
other = await login(client)
ws = await create_workspace(client, me, "我的空间")
other_ws = await create_workspace(client, other, "别人的空间")
item = await _new_item(client, me, ws["id"], title="机密方案")
resp = await client.get(f"{items_url(ws['id'])}/{item['id']}", headers=other["headers"])
assert resp.json()["code"] == 3003 # 不是成员,连工作区都进不去
listed = unwrap(
await client.get(
items_url(other_ws["id"]), params={"view": "all"}, headers=other["headers"]
)
)
assert listed["items"] == []
async def test_search_by_keyword_and_filter_by_tag(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
await _new_item(client, user, ws["id"], title="整理季度合同", tag_names=["合同", "法务"])
await _new_item(client, user, ws["id"], title="买咖啡豆")
found = unwrap(
await client.get(
items_url(ws["id"]), params={"q": "合同", "view": "all"}, headers=user["headers"]
)
)
assert [i["title"] for i in found["items"]] == ["整理季度合同"]
assert {t["name"] for t in found["items"][0]["tags"]} == {"合同", "法务"}
tagged = unwrap(
await client.get(
items_url(ws["id"]), params={"tag": "法务", "view": "all"}, headers=user["headers"]
)
)
assert [i["title"] for i in tagged["items"]] == ["整理季度合同"]
all_tags = unwrap(
await client.get(f"/api/v1/workspaces/{ws['id']}/tags", headers=user["headers"])
)
assert {t["name"] for t in all_tags["items"]} == {"合同", "法务"}
async def test_today_view_uses_local_timezone_boundaries(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
now_local = datetime.now(TZ)
today_2330 = now_local.replace(hour=23, minute=30, second=0, microsecond=0)
tomorrow_0030 = (now_local + timedelta(days=1)).replace(
hour=0, minute=30, second=0, microsecond=0
)
await _new_item(
client,
user,
ws["id"],
title="今晚交",
status="todo",
due_at=today_2330.astimezone(timezone.utc).isoformat(),
)
await _new_item(
client,
user,
ws["id"],
title="明早交",
status="todo",
due_at=tomorrow_0030.astimezone(timezone.utc).isoformat(),
)
today = unwrap(
await client.get(items_url(ws["id"]), params={"view": "today"}, headers=user["headers"])
)
assert [i["title"] for i in today["items"]] == ["今晚交"]
async def test_soft_delete_hides_from_lists_but_keeps_row(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
item = await _new_item(client, user, ws["id"], status="todo")
unwrap(
await client.delete(
f"{items_url(ws['id'])}/{item['id']}",
params={"version": item["version"]},
headers=user["headers"],
)
)
active = unwrap(
await client.get(items_url(ws["id"]), params={"view": "active"}, headers=user["headers"])
)
assert active["items"] == []
deleted = unwrap(
await client.get(items_url(ws["id"]), params={"view": "deleted"}, headers=user["headers"])
)
assert [i["id"] for i in deleted["items"]] == [item["id"]]
async def test_comments_and_activity_stream(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
item = await _new_item(client, user, ws["id"])
unwrap(
await client.post(
f"{items_url(ws['id'])}/{item['id']}/status",
json={"status": "todo", "version": item["version"]},
headers=user["headers"],
)
)
unwrap(
await client.post(
f"{items_url(ws['id'])}/{item['id']}/comments",
json={"body_md": "先找财务要模板"},
headers=user["headers"],
)
)
detail = unwrap(
await client.get(f"{items_url(ws['id'])}/{item['id']}", headers=user["headers"])
)
actions = [a["action"] for a in detail["activities"]]
assert actions[0] == "commented"
assert "status_changed" in actions
assert "created" in actions
status_change = next(a for a in detail["activities"] if a["action"] == "status_changed")
assert status_change["payload"] == {"from": "inbox", "to": "todo"}
async def test_cursor_pagination_does_not_repeat_rows(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "个人")
for index in range(3):
await _new_item(client, user, ws["id"], title=f"任务{index}", status="todo")
first = unwrap(
await client.get(
items_url(ws["id"]), params={"view": "all", "limit": 2}, headers=user["headers"]
)
)
assert len(first["items"]) == 2
assert first["has_more"] is True
assert first["next_cursor"]
second = unwrap(
await client.get(
items_url(ws["id"]),
params={"view": "all", "limit": 2, "cursor": first["next_cursor"]},
headers=user["headers"],
)
)
ids_first = {i["id"] for i in first["items"]}
ids_second = {i["id"] for i in second["items"]}
assert ids_first.isdisjoint(ids_second)
assert len(ids_second) == 1
assert second["has_more"] is False