xsl
d9249400af
12 类实体统一表 + 状态机 + version 递增 + kb_relation 递归 CTE 1~3 跳
可解释路径（防环）+ kb_color 色板（T-P2-01/05，TD-09/18）。
实现：
- packages/kb/schemas.py：12 类 EntityType 与状态 Literal 校验、
EntityCreate/Update/Out、RelationCreate、GraphNode/Hop/Path/QueryResult、
ColorOut；雪花 ID 全部字符串出参（JS 精度）。
- packages/kb/repo.py：实体 CRUD/按名查重/过滤分页搜索；traverse 递归
CTE（path 数组 + NOT @> 防环，锚点与递归段 varchar[] 显转型）；色板读写。
- packages/kb/service.py：状态机矩阵（draft→reviewing→published→archived
+ draft/reviewing→archived 撤稿，非法跃迁 40900，矩阵单测 16 组全覆盖）、
内容修改仅 draft/reviewing 且 version+1、published 只读；关系写入
（自环/三元组重复 400、端点缺失 404）；色板 role/hex(#RRGGBB) 校验。
- apps/api/routers/kb.py：GET/POST /kb/items、GET/PATCH /kb/items/{id}、
GET /kb/items/{id}/graph?max_hops=1..3（超界 422）；KbService 装配就近
放路由文件（apps/api/deps.py 本轮禁改）。
测试：单测 77 例（跃迁矩阵、12 类 Literal、非法类型拒绝、hex/role/hops
校验）；集成 5 例（真实 PG：生命周期/改 published 被拒/版本递增、关系
约束、2 跳可解释路径+防环、列表过滤分页、色板读写）。全套门禁绿：
ruff check/format、mypy（指定包列表）、lint-imports、单测 315 通过、
集成 21 通过。
偏差说明：
1. 新增 tests/unit/__init__.py 与 tests/integration/__init__.py（仅注释
的包标记）：任务文档要求单测/集成同名 test_kb_entity.py，pytest
prepend 导入模式按 basename 注册模块会撞名，加包标记唯一化。
2. 关系与色板未挂 HTTP 路由：文档 §3.4 路由清单仅列 items+graph，
录入入口属运营后台 T-P2-09，本批在 service 层落地并集成直连验证。
3. published→draft 按文档字面矩阵（其余跃迁非法）拦为 40900：改已发布
内容的既定路径是撤稿归档后重录新草稿（错误 hint 已说明），未私开
回退通道。
123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
"""kb 实体路由（FR-KB-01/02/05）：实体 CRUD + 状态机 + 1~3 跳图谱查询。
|
||
|
||
写接口暂不挂鉴权以外的前置（后台运营角色属 T-P2-09，先 ``CurrentUserId``，
|
||
``updated_by`` 记录操作者）。统一信封见架构文档 §9.2；跨域响应模型在
|
||
``packages/kb/schemas.py``（``apps/api/schemas.py`` 本轮禁改）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
|
||
from apps.api.deps import CurrentUserId, SessionDep
|
||
from packages.kb.schemas import (
|
||
MAX_GRAPH_HOPS,
|
||
EntityCreate,
|
||
EntityOut,
|
||
EntityStatus,
|
||
EntityType,
|
||
EntityUpdate,
|
||
)
|
||
from packages.kb.service import KbService
|
||
from packages.shared.models import KbEntity
|
||
from packages.shared.response import ok
|
||
|
||
router = APIRouter(prefix="/kb", tags=["kb"])
|
||
|
||
|
||
def kb_service_dep(session: SessionDep) -> KbService:
|
||
"""``apps/api/deps.py`` 本轮禁改，kb 的服务装配放路由文件内就近完成。"""
|
||
return KbService(session)
|
||
|
||
|
||
KbServiceDep = Annotated[KbService, Depends(kb_service_dep)]
|
||
|
||
|
||
def _serialize(entity: KbEntity) -> dict[str, object]:
|
||
return EntityOut(
|
||
entity_id=str(entity.entity_id),
|
||
entity_type=entity.entity_type,
|
||
name=entity.name,
|
||
attrs=dict(entity.attrs),
|
||
status=entity.status,
|
||
version=entity.version,
|
||
source=entity.source,
|
||
license=entity.license,
|
||
updated_by=entity.updated_by,
|
||
created_at=entity.created_at,
|
||
updated_at=entity.updated_at,
|
||
).model_dump()
|
||
|
||
|
||
@router.get("/items", summary="实体列表：entity_type / status 过滤 + 名称模糊 + 分页")
|
||
async def list_items(
|
||
user_id: CurrentUserId,
|
||
kb: KbServiceDep,
|
||
entity_type: EntityType | None = Query(default=None, description="12 类本体之一"),
|
||
status: EntityStatus | None = Query(default=None),
|
||
name: str | None = Query(default=None, max_length=128, description="名称模糊匹配"),
|
||
limit: int = Query(default=20, ge=1, le=100),
|
||
offset: int = Query(default=0, ge=0),
|
||
) -> dict[str, object]:
|
||
rows, total = await kb.search_entities(
|
||
entity_type=entity_type,
|
||
status=status,
|
||
name_ilike=name,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
return ok(
|
||
{
|
||
"items": [_serialize(row) for row in rows],
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset,
|
||
}
|
||
)
|
||
|
||
|
||
@router.post("/items", summary="新建实体（draft / version=1 起步）")
|
||
async def create_entity(
|
||
payload: EntityCreate,
|
||
user_id: CurrentUserId,
|
||
kb: KbServiceDep,
|
||
) -> dict[str, object]:
|
||
entity = await kb.create_entity(payload=payload, updated_by=str(user_id))
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.get("/items/{entity_id}", summary="实体详情")
|
||
async def get_entity(
|
||
entity_id: int,
|
||
user_id: CurrentUserId,
|
||
kb: KbServiceDep,
|
||
) -> dict[str, object]:
|
||
entity = await kb.get_entity(entity_id=entity_id)
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.patch("/items/{entity_id}", summary="修改内容（draft/reviewing）或推进状态（状态机校验）")
|
||
async def update_entity(
|
||
entity_id: int,
|
||
payload: EntityUpdate,
|
||
user_id: CurrentUserId,
|
||
kb: KbServiceDep,
|
||
) -> dict[str, object]:
|
||
entity = await kb.update_entity(entity_id=entity_id, payload=payload, updated_by=str(user_id))
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.get("/items/{entity_id}/graph", summary="从该实体出发的 1~3 跳可解释路径（T-P2-05）")
|
||
async def get_entity_graph(
|
||
entity_id: int,
|
||
user_id: CurrentUserId,
|
||
kb: KbServiceDep,
|
||
max_hops: int = Query(
|
||
default=MAX_GRAPH_HOPS, ge=1, le=MAX_GRAPH_HOPS, description="遍历深度 1~3，超界 422"
|
||
),
|
||
) -> dict[str, object]:
|
||
result = await kb.traverse_graph(entity_id=entity_id, max_hops=max_hops)
|
||
return ok(result.model_dump())
|