pipeline_core/wwwroot/secrets/get_user_secrets.dspy

31 lines
1.2 KiB
Plaintext

# get_user_secrets.dspy - 我的凭据列表(元数据,绝不含值/密文)
import json
uid = await get_user()
if not uid:
return []
org = (await get_userorgid()) or ''
from pipeline_service import secret_vault as sv
rows = []
async with DBPools().sqlorContext('pipeline') as sor:
rows = await sv.list_secrets(sor, org_id=org, user_id=uid, only_active=False)
await sor.sqlExe("COMMIT", {})
out = []
for r in (rows or []):
out.append({
"id": r.get("id", ""),
"name": r.get("name", ""),
"secret_type": r.get("secret_type", "") or "other",
"label": r.get("label", "") or "",
"remark": r.get("remark", "") or "",
"status": r.get("status", "") or "active",
"prefix_hint": r.get("prefix_hint", "") or "",
"length_hint": r.get("length_hint", 0) or 0,
"use_count": r.get("use_count", 0) or 0,
"source": r.get("source", "") or "",
"owner": "mine" if r.get("user_id") == uid else "org",
"created_at": str(r.get("created_at", "") or "")[:19],
"updated_at": str(r.get("updated_at", "") or "")[:19],
})
# Tabular 数据契约 = sqlPaging 同构 {total, rows}(裸 list 不渲染,实测)
return {"total": len(out), "rows": out}