feat: feature_capability.py 功能/需求状态机能力(CAS迁移+审计,scope=project_id)
This commit is contained in:
ymq
parent
7fa501d359
commit
df6f5cce34
197
pipeline_service/feature_capability.py
Normal file
197
pipeline_service/feature_capability.py
Normal file
@ -0,0 +1,197 @@
|
||||
"""功能/需求能力 — sd_features 的状态机语义化迁移（通用、产线无关）。
|
||||
|
||||
定位：本模块是功能（feature/需求）的状态机语义化操作层。CRUD（新增/编辑字段）走
|
||||
xls2ui 生成的 CRUD 端点；本模块只做「状态流转」——每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
|
||||
|
||||
功能的状态机/流转规则在 feature skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
|
||||
|
||||
角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
|
||||
|
||||
scope 约定：sd_features 用 project_id 做范围校验（CAS 的 WHERE 里带 project_id），
|
||||
与 pipeline_tasks/pipeline_agent_questions 用 tenant_id 不同——sd_* 业务表用自己的归属列。
|
||||
"""
|
||||
|
||||
import logging
|
||||
from sqlor.dbpools import DBPools
|
||||
from appPublic.uniqueID import getID
|
||||
from .audit import record_audit
|
||||
|
||||
DBNAME = "pipeline"
|
||||
logger = logging.getLogger("pipeline.feature_capability")
|
||||
|
||||
TABLE = "sd_features"
|
||||
|
||||
# 功能状态（SDLC 默认，状态机语义见 feature skill）
|
||||
S_PROPOSED = "proposed" # 已提出（待审批）
|
||||
S_APPROVED = "approved" # 已批准（待开发）
|
||||
S_IN_PROGRESS = "in_progress" # 开发中
|
||||
S_DELIVERED = "delivered" # 已交付（待验证）
|
||||
S_VERIFIED = "verified" # 已验证
|
||||
S_REJECTED = "rejected" # 已驳回
|
||||
|
||||
|
||||
def _get_db():
|
||||
db = DBPools()
|
||||
if not db.databases:
|
||||
from appPublic.jsonConfig import getConfig
|
||||
config = getConfig()
|
||||
if config.databases:
|
||||
db.databases = config.databases
|
||||
return db, DBNAME
|
||||
|
||||
|
||||
def _normalize_role(role):
|
||||
"""角色规范：agent 角色补 agent. 前缀；人角色 {orgtype}.{role} 保留原样。"""
|
||||
role = (role or "").strip()
|
||||
if not role:
|
||||
return ""
|
||||
if "." in role:
|
||||
return role
|
||||
return f"agent.{role}"
|
||||
|
||||
|
||||
async def _transition(feature_id, project_id, from_state, to_state, action,
|
||||
who=None, agent_id=None, detail=None):
|
||||
"""CAS 状态迁移 + 审计。返回 (ok, message)。"""
|
||||
if not feature_id or not project_id:
|
||||
return False, "缺少 feature_id 或 project_id"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
await sor.sqlExe(
|
||||
f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW() "
|
||||
"WHERE id=${fid}$ AND project_id=${pid}$ AND status=${from}$",
|
||||
{"to": to_state, "fid": feature_id, "pid": project_id, "from": from_state})
|
||||
recs = await sor.R(TABLE, {'id': feature_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
return False, "功能不存在"
|
||||
cur = getattr(recs[0], 'status', '')
|
||||
if cur != to_state:
|
||||
return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 status={cur}"
|
||||
await record_audit(project_id, TABLE, feature_id, action,
|
||||
from_state=from_state, to_state=to_state,
|
||||
who=who, agent_id=agent_id, detail=detail, sor=sor)
|
||||
return True, to_state
|
||||
|
||||
|
||||
async def propose_feature(project_id, feature_name, description="",
|
||||
iteration_id="", priority="P2", feature_type="new_feature",
|
||||
created_by="", who=None, agent_id=None):
|
||||
"""提出功能：新建记录，status=proposed。返回 (ok, feature_id_or_message)。"""
|
||||
if not project_id:
|
||||
return False, "缺少 project_id"
|
||||
if not feature_name or not feature_name.strip():
|
||||
return False, "缺少 feature_name"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
fid = getID()
|
||||
await sor.C(TABLE, {
|
||||
'id': fid,
|
||||
'project_id': project_id,
|
||||
'iteration_id': iteration_id or '',
|
||||
'feature_name': feature_name.strip(),
|
||||
'description': description or '',
|
||||
'feature_type': feature_type or 'new_feature',
|
||||
'priority': priority or 'P2',
|
||||
'status': S_PROPOSED,
|
||||
'created_by': created_by or '',
|
||||
})
|
||||
await record_audit(project_id, TABLE, fid, 'propose',
|
||||
to_state=S_PROPOSED, who=_normalize_role(who),
|
||||
agent_id=agent_id, sor=sor)
|
||||
logger.info("propose_feature: %s project=%s", fid, project_id)
|
||||
return True, fid
|
||||
|
||||
|
||||
async def approve_feature(feature_id, project_id, who=None, agent_id=None, comment=None):
|
||||
"""审批通过：proposed → approved。"""
|
||||
return await _transition(feature_id, project_id, S_PROPOSED, S_APPROVED, 'approve',
|
||||
who=who, agent_id=agent_id, detail=comment)
|
||||
|
||||
|
||||
async def reject_feature(feature_id, project_id, who=None, agent_id=None, comment=None):
|
||||
"""驳回：proposed → rejected（也兼容 approved → rejected）。"""
|
||||
ok, msg = await _transition(feature_id, project_id, S_PROPOSED, S_REJECTED, 'reject',
|
||||
who=who, agent_id=agent_id, detail=comment)
|
||||
if ok:
|
||||
return ok, msg
|
||||
return await _transition(feature_id, project_id, S_APPROVED, S_REJECTED, 'reject',
|
||||
who=who, agent_id=agent_id, detail=comment)
|
||||
|
||||
|
||||
async def start_feature(feature_id, project_id, who=None, agent_id=None):
|
||||
"""开始开发：approved → in_progress。"""
|
||||
return await _transition(feature_id, project_id, S_APPROVED, S_IN_PROGRESS, 'start',
|
||||
who=who, agent_id=agent_id)
|
||||
|
||||
|
||||
async def deliver_feature(feature_id, project_id, who=None, agent_id=None):
|
||||
"""交付：in_progress → delivered。"""
|
||||
return await _transition(feature_id, project_id, S_IN_PROGRESS, S_DELIVERED, 'deliver',
|
||||
who=who, agent_id=agent_id)
|
||||
|
||||
|
||||
async def verify_feature(feature_id, project_id, who=None, agent_id=None):
|
||||
"""验证通过：delivered → verified。"""
|
||||
return await _transition(feature_id, project_id, S_DELIVERED, S_VERIFIED, 'verify',
|
||||
who=who, agent_id=agent_id)
|
||||
|
||||
|
||||
async def reopen_feature(feature_id, project_id, who=None, agent_id=None, comment=None):
|
||||
"""重新打开：rejected → proposed（也兼容 verified → proposed 回炉）。"""
|
||||
ok, msg = await _transition(feature_id, project_id, S_REJECTED, S_PROPOSED, 'reopen',
|
||||
who=who, agent_id=agent_id, detail=comment)
|
||||
if ok:
|
||||
return ok, msg
|
||||
return await _transition(feature_id, project_id, S_VERIFIED, S_PROPOSED, 'reopen',
|
||||
who=who, agent_id=agent_id, detail=comment)
|
||||
|
||||
|
||||
async def set_feature_state(feature_id, project_id, from_state, to_state,
|
||||
who=None, agent_id=None, detail=None):
|
||||
"""通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
|
||||
return await _transition(feature_id, project_id, from_state, to_state, 'set_state',
|
||||
who=who, agent_id=agent_id, detail=detail)
|
||||
|
||||
|
||||
async def list_features(project_id, status=None, iteration_id=None, limit=50) -> list:
|
||||
"""列出功能（可按状态/迭代过滤）。"""
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
conditions = ["project_id=${pid}$"]
|
||||
params = {"pid": project_id}
|
||||
if status:
|
||||
conditions.append("status=${status}$")
|
||||
params["status"] = status
|
||||
if iteration_id:
|
||||
conditions.append("iteration_id=${iid}$")
|
||||
params["iid"] = iteration_id
|
||||
where = " AND ".join(conditions)
|
||||
try:
|
||||
limit = int(limit)
|
||||
except (TypeError, ValueError):
|
||||
limit = 50
|
||||
sql = (f"SELECT * FROM {TABLE} WHERE {where} "
|
||||
f"ORDER BY created_at DESC LIMIT {limit}")
|
||||
recs = await sor.sqlExe(sql, params)
|
||||
# 释放 SELECT 元数据锁
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
result = []
|
||||
for rec in (recs or []):
|
||||
result.append(_rec_to_dict(rec))
|
||||
return result
|
||||
|
||||
|
||||
def _rec_to_dict(rec):
|
||||
"""把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
|
||||
if isinstance(rec, dict):
|
||||
return dict(rec)
|
||||
try:
|
||||
return dict(rec)
|
||||
except (TypeError, ValueError):
|
||||
if hasattr(rec, 'to_dict'):
|
||||
try:
|
||||
return rec.to_dict()
|
||||
except Exception:
|
||||
pass
|
||||
return {}
|
||||
Loading…
x
Reference in New Issue
Block a user