YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
wangjiming
2026-08-03 09:34:08 +08:00
parent b975de02da
commit 15c4223f2c
43 changed files with 4498 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

746
backend/app/db/platform_store.py
View File

@@ -14,6 +14,7 @@ from pathlib import Path
from typing import Any, Iterator
import psycopg
from psycopg_pool import ConnectionPool
from app.core.config import get_settings
@@ -291,21 +292,54 @@ class PlatformStore:
def __init__(self, database_url: str | None = None) -> None:
settings = get_settings()
self.database_url = _psycopg_url(database_url or settings.database_url)
# Reuse connections via a pool to avoid the TCP+auth handshake on every
# request (notably expensive against the remote PostgreSQL instance).
# TCP keepalive 让操作系统持续保活连接，抵抗远程库空闲静默断连。
pool_kwargs = {
"keepalives": 1,
"keepalives_idle": 30,
"keepalives_interval": 10,
"keepalives_count": 5,
}
self._pool = ConnectionPool(
conninfo=self.database_url,
kwargs=pool_kwargs,
min_size=2,
max_size=10,
# 借出前校验连接可用性，避免执行 SQL 时才发现 [BAD] 再重建。
check=ConnectionPool.check_connection,
# 不主动回收空闲连接（远程库约 10s 断，由 keepalive 维持），
# 减少无谓的重建握手。
max_idle=0,
# 请求最多排队等待 5s，避免雪崩时无限堆积。
max_waiting=16,
open=False,
)
# 注意：不要在此调用 pool.wait()，它会阻塞等待 min_size 个连接就绪，
# 在远程库响应慢/超时时会卡死 uvicorn worker 进程，导致所有请求无响应。
self._pool.open()
self.ensure_schema()
self.ensure_seed_data()
@contextmanager
def connect(self) -> Iterator["PgConnection"]:
raw_conn = psycopg.connect(self.database_url)
conn = PgConnection(raw_conn)
with self._pool.connection() as raw_conn:
conn = PgConnection(raw_conn)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def close_pool(self) -> None:
"""Release pooled connections. Safe to call multiple times."""
try:
yield conn
conn.commit()
self._pool.close()
except Exception:
conn.rollback()
raise
finally:
conn.close()
pass
def ensure_schema(self) -> None:
schema_path = Path(__file__).with_name("sql") / "001_platform_runtime.sql"
@@ -335,6 +369,11 @@ class PlatformStore:
"last_error": "TEXT",
},
)
schema_dir = Path(__file__).with_name("sql")
for extra in ("002_governance.sql", "003_tenant_quota.sql"):
extra_path = schema_dir / extra
if extra_path.exists():
conn.executescript(extra_path.read_text(encoding="utf-8"))
def _column_names(self, conn: PgConnection, table_name: str) -> set[str]:
columns = conn.execute(
@@ -1095,6 +1134,18 @@ class PlatformStore:
raise ValueError("protected user cannot be deleted")
conn.execute("DELETE FROM users WHERE id=?", (user_id,))
def reset_password(self, user_id: str, new_password: str) -> None:
with self.connect() as conn:
row = conn.execute("SELECT protected FROM users WHERE id=?", (user_id,)).fetchone()
if not row:
raise KeyError(user_id)
if row["protected"]:
raise ValueError("protected user cannot reset password")
conn.execute(
"UPDATE users SET password_hash=? WHERE id=?",
(hash_password(new_password), user_id),
)
def _user(self, row: PgRow) -> dict[str, Any]:
return {
"id": row["id"],
@@ -2891,6 +2942,672 @@ class PlatformStore:
]
return {"file": file_name, "content": "\n".join(lines), "size": "1 KB"}
# ===================== 平台治理：角色 =====================
def roles(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM roles ORDER BY name").fetchall()
return [dict(r) for r in rows]
# ===================== 平台治理：审计日志 =====================
def audit_logs(
self,
*,
tenant_id: str | None = None,
project_id: str | None = None,
actor_id: str | None = None,
action: str | None = None,
target_type: str | None = None,
start_time: str | None = None,
end_time: str | None = None,
limit: int = 50,
offset: int = 0,
) -> dict[str, Any]:
clauses: list[str] = []
params: list[Any] = []
if tenant_id:
clauses.append("tenant_id=?")
params.append(tenant_id)
if project_id:
clauses.append("project_id=?")
params.append(project_id)
if actor_id:
clauses.append("actor_id=?")
params.append(actor_id)
if action:
clauses.append("action=?")
params.append(action)
if target_type:
clauses.append("target_type=?")
params.append(target_type)
if start_time:
clauses.append("time>=?")
params.append(start_time)
if end_time:
clauses.append("time<=?")
params.append(end_time)
where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
with self.connect() as conn:
total = conn.execute(f"SELECT COUNT(*) AS c FROM audit_logs{where}", tuple(params)).fetchone()["c"]
params_paged = list(params) + [limit, offset]
rows = conn.execute(
f"SELECT * FROM audit_logs{where} ORDER BY time DESC LIMIT ? OFFSET ?",
tuple(params_paged),
).fetchall()
return {"total": total, "items": [dict(r) for r in rows]}
def record_audit(
self,
*,
action: str,
actor_id: str | None = None,
target_type: str | None = None,
target_id: str | None = None,
tenant_id: str | None = None,
project_id: str | None = None,
detail: str | None = None,
ip: str | None = None,
) -> None:
with self.connect() as conn:
conn.execute(
"""
INSERT INTO audit_logs
(id, tenant_id, project_id, actor_id, action, target_type, target_id, detail, client_ip, time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
new_id("log"),
tenant_id,
project_id,
actor_id,
action,
target_type,
target_id,
detail,
ip,
utcnow(),
),
)
# ===================== 平台治理：会话 =====================
def create_session(self, user_id: str, *, ip: str | None = None) -> dict[str, Any]:
sid = new_id("sess")
login_at = utcnow()
with self.connect() as conn:
conn.execute(
"INSERT INTO sessions (id, user_id, username, login_at, create_time) "
"VALUES (%s, %s, (SELECT username FROM users WHERE id=%s), %s, %s)",
(sid, user_id, user_id, login_at, login_at),
)
return {"session_id": sid, "user_id": user_id, "login_at": login_at}
def finish_session(self, session_id: str) -> None:
"""登出时记录 logout_at 与时长（秒）。"""
logout_at = utcnow()
with self.connect() as conn:
conn.execute(
"UPDATE sessions SET logout_at=%s, "
"duration_seconds=EXTRACT(EPOCH FROM (%s::timestamptz - login_at::timestamptz))::int "
"WHERE id=%s AND logout_at IS NULL",
(logout_at, logout_at, session_id),
)
def active_sessions(self, user_id: str) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute(
"SELECT * FROM sessions WHERE user_id=%s AND logout_at IS NULL "
"ORDER BY login_at DESC",
(user_id,),
).fetchall()
return [dict(r) for r in rows]
def destroy_session(self, session_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM sessions WHERE id=%s", (session_id,))
def extend_session(self, session_id: str, *, expires_in_seconds: int = 3600 * 8) -> dict[str, Any] | None:
# 兼容旧调用，仅更新 login_at 之后延长的含义在此简化为 no-op 返回现有记录。
with self.connect() as conn:
row = conn.execute("SELECT * FROM sessions WHERE id=%s", (session_id,)).fetchone()
if not row:
return None
return {
"session_id": session_id,
"user_id": row["user_id"],
"login_at": row["login_at"],
}
def set_session_user(self, session_id: str, user_id: str) -> None:
with self.connect() as conn:
conn.execute("UPDATE sessions SET user_id=%s WHERE id=%s", (user_id, session_id))
def login_duration_rank(self, limit: int = 8, days: int = 30) -> list[dict[str, Any]]:
"""登录时长排行：按用户聚合近 N 天的会话时长（小时）。
sessions 表列：login_at(TEXT), logout_at(TEXT), duration_seconds(INT)。
优先用 duration_seconds；为空时回退计算 now-login_at（未登出）或 logout_at-login_at。
"""
with self.connect() as conn:
rows = conn.execute(
"SELECT s.user_id, s.login_at, s.logout_at, s.duration_seconds, "
"u.username, u.display_name, u.role "
"FROM sessions s LEFT JOIN users u ON s.user_id = u.id "
"WHERE s.login_at::timestamptz >= NOW() - make_interval(days => %s)",
(days,),
).fetchall()
now = datetime.now(timezone.utc)
agg: dict[str, dict[str, Any]] = {}
for r in rows:
uid = r["user_id"] or ""
bucket = agg.setdefault(
uid,
{
"user": r["display_name"] or r["username"] or uid,
"role": r["role"] or "",
"total": 0.0,
},
)
dur = r["duration_seconds"]
if dur is not None:
bucket["total"] += float(dur)
continue
start = parse_time(r["login_at"])
end = parse_time(r["logout_at"]) if r["logout_at"] else None
if start and end:
bucket["total"] += max(0, (end - start).total_seconds())
elif start:
bucket["total"] += max(0, (now - start).total_seconds())
result = [
{"user": b["user"], "role": b["role"], "duration": round(b["total"] / 3600, 1)}
for b in agg.values()
]
result.sort(key=lambda x: x["duration"], reverse=True)
return result[:limit]
# ===================== 平台治理：审批 =====================
def create_approval_template(self, payload: dict[str, Any]) -> dict[str, Any]:
with self.connect() as conn:
tid = new_id("tpl")
conn.execute(
"INSERT INTO approval_templates (id, name, steps, create_time) VALUES (?, ?, ?, ?)",
(tid, payload["name"], json_dumps(payload.get("steps", [])), utcnow()),
)
return self.approval_template(tid)
def approval_templates(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM approval_templates ORDER BY create_time DESC").fetchall()
return [dict(r) for r in rows]
def approval_template(self, template_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM approval_templates WHERE id=?", (template_id,)).fetchone()
if not row:
raise KeyError(template_id)
return dict(row)
def update_approval_template(self, template_id: str, payload: dict[str, Any]) -> dict[str, Any]:
fields = {k: v for k, v in payload.items() if k in ("name", "steps")}
if "steps" in fields:
fields["steps"] = json_dumps(fields["steps"])
if not fields:
return self.approval_template(template_id)
set_clause = ", ".join(f"{k}=?" for k in fields)
params = list(fields.values()) + [template_id]
with self.connect() as conn:
conn.execute(f"UPDATE approval_templates SET {set_clause} WHERE id=?", tuple(params))
return self.approval_template(template_id)
def delete_approval_template(self, template_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM approval_templates WHERE id=?", (template_id,)).fetchone()
if not row:
raise KeyError(template_id)
conn.execute("DELETE FROM approval_templates WHERE id=?", (template_id,))
return dict(row)
def create_approval_instance(self, payload: dict[str, Any]) -> dict[str, Any]:
template_id = payload.get("template_id")
steps = []
if template_id:
tpl = self.approval_template(template_id)
steps = json_loads(tpl["steps"]) if tpl.get("steps") else []
with self.connect() as conn:
iid = new_id("appr")
conn.execute(
"""
INSERT INTO approval_instances
(id, template_id, resource_type, resource_id, applicant_id, status, current_step, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
iid,
template_id,
payload["resource_type"],
payload["resource_id"],
payload["applicant_id"],
"pending",
0,
utcnow(),
),
)
for idx, step in enumerate(steps):
conn.execute(
"INSERT INTO approval_steps (id, instance_id, step_index, approver_id, status, time) VALUES (?, ?, ?, ?, ?, ?)",
(new_id("step"), iid, idx, step.get("approver_id"), "pending", None),
)
return self.approval_instance(iid)
def approval_instances(self, *, status: str | None = None) -> list[dict[str, Any]]:
with self.connect() as conn:
if status:
rows = conn.execute(
"SELECT * FROM approval_instances WHERE status=? ORDER BY create_time DESC", (status,)
).fetchall()
else:
rows = conn.execute("SELECT * FROM approval_instances ORDER BY create_time DESC").fetchall()
return [dict(r) for r in rows]
def approval_instance(self, instance_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM approval_instances WHERE id=?", (instance_id,)).fetchone()
if not row:
raise KeyError(instance_id)
steps = conn.execute(
"SELECT * FROM approval_steps WHERE instance_id=? ORDER BY step_index", (instance_id,)
).fetchall()
result = dict(row)
result["steps"] = [dict(s) for s in steps]
return result
def decide_approval_step(self, instance_id: str, step_index: int, *, approver_id: str, approved: bool, comment: str | None = None) -> dict[str, Any]:
with self.connect() as conn:
inst = conn.execute("SELECT * FROM approval_instances WHERE id=?", (instance_id,)).fetchone()
if not inst:
raise KeyError(instance_id)
if inst["status"] != "pending":
raise ValueError("instance not pending")
step = conn.execute(
"SELECT * FROM approval_steps WHERE instance_id=? AND step_index=?",
(instance_id, step_index),
).fetchone()
if not step:
raise KeyError("step not found")
if step["status"] != "pending":
raise ValueError("step already decided")
new_status = "approved" if approved else "rejected"
conn.execute(
"UPDATE approval_steps SET status=?, comment=?, time=? WHERE id=?",
(new_status, comment, utcnow(), step["id"]),
)
if approved:
conn.execute(
"UPDATE approval_instances SET current_step=? WHERE id=?",
(step_index + 1, instance_id),
)
step_rows = conn.execute(
"SELECT * FROM approval_steps WHERE instance_id=? ORDER BY step_index", (instance_id,)
).fetchall()
if all(s["status"] == "approved" for s in step_rows):
conn.execute("UPDATE approval_instances SET status='approved' WHERE id=?", (instance_id,))
else:
conn.execute("UPDATE approval_instances SET status='rejected' WHERE id=?", (instance_id,))
return self.approval_instance(instance_id)
# ===================== 平台治理：租户 =====================
def tenants(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM tenants ORDER BY create_time DESC").fetchall()
return [dict(r) for r in rows]
def tenant(self, tenant_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM tenants WHERE id=?", (tenant_id,)).fetchone()
if not row:
raise KeyError(tenant_id)
return dict(row)
def create_tenant(self, payload: dict[str, Any]) -> dict[str, Any]:
with self.connect() as conn:
tid = new_id("tnt")
conn.execute(
"""
INSERT INTO tenants (id, name, code, status, owner_user_id, quota, retention_policy_id, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
tid,
payload["name"],
payload.get("code"),
"active",
payload.get("owner_user_id"),
json_dumps(payload.get("quota", {})),
payload.get("retention_policy_id"),
utcnow(),
),
)
return self.tenant(tid)
def update_tenant(self, tenant_id: str, payload: dict[str, Any]) -> dict[str, Any]:
fields = {k: v for k, v in payload.items() if k in ("name", "code", "status", "owner_user_id", "quota", "retention_policy_id")}
if "quota" in fields:
fields["quota"] = json_dumps(fields["quota"])
if not fields:
return self.tenant(tenant_id)
set_clause = ", ".join(f"{k}=?" for k in fields)
params = list(fields.values()) + [tenant_id]
with self.connect() as conn:
conn.execute(f"UPDATE tenants SET {set_clause} WHERE id=?", tuple(params))
return self.tenant(tenant_id)
def set_tenant_quota(self, tenant_id: str, quota: dict[str, Any]) -> dict[str, Any]:
with self.connect() as conn:
conn.execute("UPDATE tenants SET quota=? WHERE id=?", (json_dumps(quota), tenant_id))
return self.tenant(tenant_id)
def set_tenant_retention(self, tenant_id: str, retention_policy_id: str | None) -> dict[str, Any]:
with self.connect() as conn:
conn.execute("UPDATE tenants SET retention_policy_id=? WHERE id=?", (retention_policy_id, tenant_id))
return self.tenant(tenant_id)
def delete_tenant(self, tenant_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM tenants WHERE id=?", (tenant_id,)).fetchone()
if not row:
raise KeyError(tenant_id)
conn.execute("DELETE FROM tenants WHERE id=?", (tenant_id,))
return dict(row)
def get_acl(self, resource_type: str, resource_id: str) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute(
"SELECT * FROM acls WHERE resource_type=? AND resource_id=?",
(resource_type, resource_id),
).fetchall()
return [dict(r) for r in rows]
def set_acl(self, resource_type: str, resource_id: str, entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
with self.connect() as conn:
conn.execute(
"DELETE FROM acls WHERE resource_type=? AND resource_id=?",
(resource_type, resource_id),
)
for e in entries:
conn.execute(
"""
INSERT INTO acls (id, resource_type, resource_id, principal_type, principal_id, permission, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
new_id("acl"),
resource_type,
resource_id,
e.get("principal_type"),
e.get("principal_id"),
e.get("permission"),
utcnow(),
),
)
rows = conn.execute(
"SELECT * FROM acls WHERE resource_type=? AND resource_id=?",
(resource_type, resource_id),
).fetchall()
return [dict(r) for r in rows]
# ===================== 平台治理：项目空间 =====================
def projects(self, *, tenant_id: str = "default", status: str | None = None, keyword: str | None = None) -> list[dict[str, Any]]:
clauses = ["tenant_id=?"]
params: list[Any] = [tenant_id]
if status:
clauses.append("status=?")
params.append(status)
if keyword:
clauses.append("(name LIKE ? OR code LIKE ?)")
params.extend([f"%{keyword}%", f"%{keyword}%"])
with self.connect() as conn:
rows = conn.execute(
f"SELECT * FROM projects WHERE {' AND '.join(clauses)} ORDER BY create_time DESC",
tuple(params),
).fetchall()
return [dict(r) for r in rows]
def project(self, project_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM projects WHERE id=?", (project_id,)).fetchone()
if not row:
raise KeyError(project_id)
return dict(row)
def create_project(self, payload: dict[str, Any]) -> dict[str, Any]:
with self.connect() as conn:
pid = new_id("prj")
conn.execute(
"""
INSERT INTO projects (id, tenant_id, name, code, description, quota, status, create_time, create_by, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
pid,
payload.get("tenant_id", "default"),
payload["name"],
payload["code"],
payload.get("description"),
json_dumps(payload.get("quota", {})),
"active",
utcnow(),
payload.get("create_by"),
utcnow(),
),
)
return self.project(pid)
def update_project(self, project_id: str, payload: dict[str, Any]) -> dict[str, Any]:
fields = {k: v for k, v in payload.items() if k in ("name", "code", "description", "quota", "status")}
if "quota" in fields:
fields["quota"] = json_dumps(fields["quota"])
if not fields:
return self.project(project_id)
set_clause = ", ".join(f"{k}=?" for k in fields)
params = list(fields.values()) + [project_id]
with self.connect() as conn:
conn.execute(f"UPDATE projects SET {set_clause} WHERE id=?", tuple(params))
return self.project(project_id)
def archive_project(self, project_id: str) -> dict[str, Any]:
with self.connect() as conn:
conn.execute("UPDATE projects SET status='archived' WHERE id=?", (project_id,))
return self.project(project_id)
def activate_project(self, project_id: str) -> dict[str, Any]:
with self.connect() as conn:
conn.execute("UPDATE projects SET status='active' WHERE id=?", (project_id,))
return self.project(project_id)
def delete_project(self, project_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM projects WHERE id=?", (project_id,))
def project_members(self, project_id: str) -> list[dict[str, Any]]:
with self.connect() as conn:
row = conn.execute("SELECT * FROM projects WHERE id=?", (project_id,)).fetchone()
if not row:
raise KeyError(project_id)
rows = conn.execute(
"""
SELECT pm.*, u.username, u.display_name
FROM project_members pm JOIN users u ON u.id = pm.user_id
WHERE pm.project_id=?
""",
(project_id,),
).fetchall()
return [dict(r) for r in rows]
def add_project_member(self, project_id: str, payload: dict[str, Any]) -> dict[str, Any]:
user_id = payload["user_id"]
role = payload.get("role", "member")
with self.connect() as conn:
conn.execute(
"INSERT INTO project_members (project_id, user_id, role, create_time) VALUES (?, ?, ?, ?)",
(project_id, user_id, role, utcnow()),
)
row = conn.execute(
"""
SELECT pm.*, u.username, u.display_name
FROM project_members pm JOIN users u ON u.id = pm.user_id
WHERE pm.project_id=? AND pm.user_id=?
""",
(project_id, user_id),
).fetchone()
return {
"project_id": row["project_id"],
"user_id": row["user_id"],
"username": row["username"],
"display_name": row["display_name"],
"role": row["role"],
"create_time": row["create_time"],
}
def update_project_member_role(self, project_id: str, user_id: str, role: str) -> dict[str, Any]:
with self.connect() as conn:
conn.execute(
"UPDATE project_members SET role=? WHERE project_id=? AND user_id=?",
(role, project_id, user_id),
)
row = conn.execute(
"""
SELECT pm.*, u.username, u.display_name
FROM project_members pm JOIN users u ON u.id = pm.user_id
WHERE pm.project_id=? AND pm.user_id=?
""",
(project_id, user_id),
).fetchone()
if not row:
raise KeyError(user_id)
return {
"project_id": row["project_id"],
"user_id": row["user_id"],
"username": row["username"],
"display_name": row["display_name"],
"role": row["role"],
"create_time": row["create_time"],
}
def remove_project_member(self, project_id: str, user_id: str) -> None:
with self.connect() as conn:
conn.execute(
"DELETE FROM project_members WHERE project_id=? AND user_id=?",
(project_id, user_id),
)
# ===================== 平台治理：资源 ACL =====================
def resource_acl(self, resource_type: str, resource_id: str) -> list[dict[str, Any]]:
"""返回资源 ACL，按主体分组，permissions 为数组。"""
rows = self.get_acl(resource_type, resource_id)
grouped: dict[str, dict[str, Any]] = {}
for r in rows:
key = f"{r.get('principal_type')}:{r.get('principal_id')}"
bucket = grouped.setdefault(
key,
{
"subject_type": r.get("principal_type"),
"subject_id": r.get("principal_id"),
"permissions": [],
},
)
perm = r.get("permission")
if perm and perm not in bucket["permissions"]:
bucket["permissions"].append(perm)
return list(grouped.values())
def set_resource_acl(
self, resource_type: str, resource_id: str, entries: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""按前端格式设置资源 ACL：entries 为 [{subject_type, subject_id, permissions: []}]。"""
flat: list[dict[str, Any]] = []
for e in entries:
for perm in e.get("permissions") or []:
flat.append(
{
"principal_type": e.get("subject_type"),
"principal_id": e.get("subject_id"),
"permission": perm,
}
)
self.set_acl(resource_type, resource_id, flat)
return self.resource_acl(resource_type, resource_id)
# ===================== 平台治理：留存策略 =====================
def retention_policies(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute(
"SELECT * FROM retention_policies ORDER BY create_time DESC"
).fetchall()
return [dict(r) for r in rows]
def retention_policy(self, policy_id: str) -> dict[str, Any]:
with self.connect() as conn:
row = conn.execute(
"SELECT * FROM retention_policies WHERE id=?", (policy_id,)
).fetchone()
if not row:
raise KeyError(policy_id)
return dict(row)
def create_retention_policy(self, payload: dict[str, Any]) -> dict[str, Any]:
pid = payload.get("id") or new_id("rpol")
with self.connect() as conn:
conn.execute(
"""
INSERT INTO retention_policies
(id, name, scope, rule, status, create_time, create_by, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
pid,
payload["name"],
payload.get("scope"),
payload.get("rule"),
payload.get("status", "active"),
utcnow(),
payload.get("create_by"),
utcnow(),
),
)
return self.retention_policy(pid)
def update_retention_policy(
self, policy_id: str, payload: dict[str, Any]
) -> dict[str, Any]:
fields = {
k: v
for k, v in payload.items()
if k in ("name", "scope", "rule", "status")
}
if not fields:
return self.retention_policy(policy_id)
fields["updated_at"] = utcnow()
set_clause = ", ".join(f"{k}=?" for k in fields)
params = list(fields.values()) + [policy_id]
with self.connect() as conn:
conn.execute(
f"UPDATE retention_policies SET {set_clause} WHERE id=?",
tuple(params),
)
return self.retention_policy(policy_id)
def delete_retention_policy(self, policy_id: str) -> None:
with self.connect() as conn:
conn.execute(
"DELETE FROM retention_policies WHERE id=?", (policy_id,)
)
_store: PlatformStore | None = None
@@ -2900,3 +3617,16 @@ def get_platform_store() -> PlatformStore:
if _store is None:
_store = PlatformStore()
return _store
import atexit as _atexit
def _close_store_pool() -> None:
global _store
if _store is not None:
_store.close_pool()
_store = None
_atexit.register(_close_store_pool)