486 lines
17 KiB
Python
486 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
from psycopg import sql
|
|
from psycopg.types.json import Jsonb
|
|
|
|
from app.config import settings
|
|
from app.data_platform.registry import TABLE_DEFINITIONS, TABLE_REGISTRY, FieldDefinition, TableDefinition
|
|
from app.data_platform.schema import (
|
|
ensure_all_project_databases,
|
|
ensure_project_database,
|
|
get_project_database,
|
|
)
|
|
from app.db import get_conn
|
|
|
|
|
|
def _json_safe(value: Any) -> Any:
|
|
if isinstance(value, Decimal):
|
|
return float(value)
|
|
if isinstance(value, uuid.UUID):
|
|
return str(value)
|
|
if hasattr(value, "isoformat"):
|
|
return value.isoformat()
|
|
return value
|
|
|
|
|
|
def _row_json(row: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if row is None:
|
|
return None
|
|
return {key: _json_safe(value) for key, value in dict(row).items()}
|
|
|
|
|
|
def _table_or_404(table_code: str) -> TableDefinition:
|
|
table = TABLE_REGISTRY.get(table_code)
|
|
if not table:
|
|
raise HTTPException(404, "数据表未注册")
|
|
return table
|
|
|
|
|
|
async def _database_or_404(project_id: str) -> dict[str, Any]:
|
|
database = await get_project_database(project_id)
|
|
if database:
|
|
return database
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"SELECT project_id, tenant_id, display_name FROM {}.projects "
|
|
"WHERE project_id=%s AND status <> 'archived'"
|
|
).format(sql.Identifier(settings.db_schema)),
|
|
(project_id,),
|
|
)
|
|
project = await cur.fetchone()
|
|
if not project:
|
|
raise HTTPException(404, "项目数据库不存在")
|
|
return await ensure_project_database(
|
|
str(project["project_id"]),
|
|
str(project["tenant_id"]),
|
|
str(project["display_name"]),
|
|
)
|
|
|
|
|
|
def _coerce_value(field: FieldDefinition, value: Any) -> Any:
|
|
if value in ("", None):
|
|
return None
|
|
if field.data_type == "json":
|
|
if isinstance(value, str):
|
|
try:
|
|
value = json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(400, f"{field.label}不是合法 JSON") from exc
|
|
return Jsonb(value)
|
|
if field.data_type == "boolean":
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).lower() in {"true", "1", "yes", "是"}
|
|
if field.data_type == "number":
|
|
if "INTEGER" in field.sql_type:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise HTTPException(400, f"{field.label}必须是整数") from exc
|
|
try:
|
|
return Decimal(str(value))
|
|
except Exception as exc:
|
|
raise HTTPException(400, f"{field.label}必须是数字") from exc
|
|
return value
|
|
|
|
|
|
def _validated_payload(
|
|
table: TableDefinition,
|
|
body: dict[str, Any],
|
|
*,
|
|
create: bool,
|
|
) -> dict[str, Any]:
|
|
definitions = {field.code: field for field in table.fields}
|
|
unknown = sorted(set(body) - set(definitions))
|
|
if unknown:
|
|
raise HTTPException(400, f"不允许的字段：{', '.join(unknown)}")
|
|
payload: dict[str, Any] = {}
|
|
for code, value in body.items():
|
|
field = definitions[code]
|
|
if not field.editable:
|
|
continue
|
|
payload[code] = _coerce_value(field, value)
|
|
if create:
|
|
missing = [
|
|
field.label
|
|
for field in table.fields
|
|
if field.required and payload.get(field.code) in (None, "")
|
|
]
|
|
if missing:
|
|
raise HTTPException(400, f"缺少必填字段：{', '.join(missing)}")
|
|
return payload
|
|
|
|
|
|
async def list_databases() -> list[dict[str, Any]]:
|
|
await ensure_all_project_databases()
|
|
admin_schema = sql.Identifier(settings.db_schema)
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"""
|
|
SELECT d.*, p.status AS project_status
|
|
FROM {}.project_databases d
|
|
JOIN {}.projects p ON p.project_id=d.project_id
|
|
WHERE p.status <> 'archived'
|
|
ORDER BY
|
|
CASE WHEN d.project_id='yunyou_libo' THEN 0 ELSE 1 END,
|
|
p.created_at DESC
|
|
"""
|
|
).format(admin_schema, admin_schema)
|
|
)
|
|
databases = await cur.fetchall()
|
|
result: list[dict[str, Any]] = []
|
|
for database in databases:
|
|
schema_name = str(database["schema_name"])
|
|
table_counts: list[int] = []
|
|
latest_updates: list[Any] = []
|
|
for table in TABLE_DEFINITIONS:
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"SELECT count(*) AS count, max(updated_at) AS updated_at "
|
|
"FROM {}.{} WHERE deleted_at IS NULL"
|
|
).format(
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
)
|
|
)
|
|
summary = await cur.fetchone()
|
|
table_counts.append(int(summary["count"] or 0))
|
|
if summary["updated_at"]:
|
|
latest_updates.append(summary["updated_at"])
|
|
row = _row_json(dict(database)) or {}
|
|
row.update(
|
|
{
|
|
"table_count": len(TABLE_DEFINITIONS),
|
|
"record_count": sum(table_counts),
|
|
"updated_at": _json_safe(max(latest_updates)) if latest_updates else _json_safe(database["updated_at"]),
|
|
}
|
|
)
|
|
result.append(row)
|
|
return result
|
|
|
|
|
|
async def list_tables(project_id: str) -> dict[str, Any]:
|
|
database = await _database_or_404(project_id)
|
|
schema_name = str(database["schema_name"])
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
tables: list[dict[str, Any]] = []
|
|
for definition in TABLE_DEFINITIONS:
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"SELECT count(*) AS count, max(updated_at) AS updated_at "
|
|
"FROM {}.{} WHERE deleted_at IS NULL"
|
|
).format(
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(definition.code),
|
|
)
|
|
)
|
|
summary = await cur.fetchone()
|
|
item = definition.as_dict()
|
|
item.update(
|
|
{
|
|
"record_count": int(summary["count"] or 0),
|
|
"updated_at": _json_safe(summary["updated_at"]),
|
|
}
|
|
)
|
|
tables.append(item)
|
|
return {"database": _row_json(database), "tables": tables}
|
|
|
|
|
|
async def list_records(
|
|
project_id: str,
|
|
table_code: str,
|
|
*,
|
|
page: int,
|
|
page_size: int,
|
|
search: str | None,
|
|
sort_field: str | None,
|
|
sort_order: str,
|
|
) -> dict[str, Any]:
|
|
table = _table_or_404(table_code)
|
|
database = await _database_or_404(project_id)
|
|
schema_name = str(database["schema_name"])
|
|
page = max(1, page)
|
|
page_size = max(1, min(200, page_size))
|
|
offset = (page - 1) * page_size
|
|
allowed_sort = {"id", "created_at", "updated_at", *(field.code for field in table.fields if field.sortable)}
|
|
order_field = sort_field if sort_field in allowed_sort else "updated_at"
|
|
order_keyword = sql.SQL("ASC") if sort_order.lower() == "asc" else sql.SQL("DESC")
|
|
|
|
where_parts: list[sql.Composable] = [sql.SQL("deleted_at IS NULL")]
|
|
params: list[Any] = []
|
|
searchable = [field for field in table.fields if field.searchable]
|
|
if search and searchable:
|
|
pattern = f"%{search.strip()}%"
|
|
where_parts.append(
|
|
sql.SQL("(")
|
|
+ sql.SQL(" OR ").join(
|
|
sql.SQL("{}::text ILIKE %s").format(sql.Identifier(field.code))
|
|
for field in searchable
|
|
)
|
|
+ sql.SQL(")")
|
|
)
|
|
params.extend(pattern for _ in searchable)
|
|
where_clause = sql.SQL(" AND ").join(where_parts)
|
|
columns = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
|
|
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
sql.SQL("SELECT count(*) AS count FROM {}.{} WHERE {}").format(
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
where_clause,
|
|
),
|
|
params,
|
|
)
|
|
total = int((await cur.fetchone())["count"])
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"SELECT {} FROM {}.{} WHERE {} ORDER BY {} {} LIMIT %s OFFSET %s"
|
|
).format(
|
|
sql.SQL(", ").join(sql.Identifier(column) for column in columns),
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
where_clause,
|
|
sql.Identifier(order_field),
|
|
order_keyword,
|
|
),
|
|
[*params, page_size, offset],
|
|
)
|
|
rows = [_row_json(dict(row)) for row in await cur.fetchall()]
|
|
return {
|
|
"table": table.as_dict(),
|
|
"items": rows,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total": total,
|
|
}
|
|
|
|
|
|
async def _write_audit(
|
|
cur,
|
|
*,
|
|
schema_name: str,
|
|
tenant_id: str,
|
|
project_id: str,
|
|
table_code: str,
|
|
record_id: uuid.UUID,
|
|
operation: str,
|
|
before_data: dict[str, Any] | None,
|
|
after_data: dict[str, Any] | None,
|
|
actor: str,
|
|
) -> None:
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"""
|
|
INSERT INTO {}.data_change_logs (
|
|
tenant_id, project_id, table_code, record_id, operation,
|
|
before_data, after_data, actor
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
).format(sql.Identifier(schema_name)),
|
|
(
|
|
tenant_id,
|
|
project_id,
|
|
table_code,
|
|
record_id,
|
|
operation,
|
|
Jsonb(before_data) if before_data is not None else None,
|
|
Jsonb(after_data) if after_data is not None else None,
|
|
actor,
|
|
),
|
|
)
|
|
record_version = int((after_data or before_data or {}).get("version") or 1)
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"""
|
|
INSERT INTO {}.graph_sync_queue (
|
|
tenant_id, project_id, table_code, record_id, operation, record_version
|
|
) VALUES (%s, %s, %s, %s, %s, %s)
|
|
"""
|
|
).format(sql.Identifier(schema_name)),
|
|
(tenant_id, project_id, table_code, record_id, operation, record_version),
|
|
)
|
|
|
|
|
|
async def create_record(
|
|
project_id: str,
|
|
table_code: str,
|
|
body: dict[str, Any],
|
|
actor: str,
|
|
) -> dict[str, Any]:
|
|
table = _table_or_404(table_code)
|
|
if not table.allow_create:
|
|
raise HTTPException(403, "该表不允许新增")
|
|
database = await _database_or_404(project_id)
|
|
payload = _validated_payload(table, body, create=True)
|
|
record_id = uuid.uuid4()
|
|
schema_name = str(database["schema_name"])
|
|
tenant_id = str(database["tenant_id"])
|
|
columns = ["id", "tenant_id", "project_id", *payload.keys()]
|
|
values = [record_id, tenant_id, project_id, *payload.values()]
|
|
returning = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
|
|
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
sql.SQL("INSERT INTO {}.{} ({}) VALUES ({}) RETURNING {}").format(
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
sql.SQL(", ").join(sql.Identifier(column) for column in columns),
|
|
sql.SQL(", ").join(sql.Placeholder() for _ in values),
|
|
sql.SQL(", ").join(sql.Identifier(column) for column in returning),
|
|
),
|
|
values,
|
|
)
|
|
row = _row_json(dict(await cur.fetchone()))
|
|
await _write_audit(
|
|
cur,
|
|
schema_name=schema_name,
|
|
tenant_id=tenant_id,
|
|
project_id=project_id,
|
|
table_code=table_code,
|
|
record_id=record_id,
|
|
operation="create",
|
|
before_data=None,
|
|
after_data=row,
|
|
actor=actor,
|
|
)
|
|
await conn.commit()
|
|
return row or {}
|
|
|
|
|
|
async def update_record(
|
|
project_id: str,
|
|
table_code: str,
|
|
record_id: str,
|
|
body: dict[str, Any],
|
|
actor: str,
|
|
) -> dict[str, Any]:
|
|
table = _table_or_404(table_code)
|
|
if not table.allow_update:
|
|
raise HTTPException(403, "该表不允许修改")
|
|
try:
|
|
record_uuid = uuid.UUID(record_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, "记录 ID 格式错误") from exc
|
|
database = await _database_or_404(project_id)
|
|
payload = _validated_payload(table, body, create=False)
|
|
if not payload:
|
|
raise HTTPException(400, "没有可修改字段")
|
|
schema_name = str(database["schema_name"])
|
|
tenant_id = str(database["tenant_id"])
|
|
returning = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
|
|
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
sql.SQL("SELECT {} FROM {}.{} WHERE id=%s AND deleted_at IS NULL").format(
|
|
sql.SQL(", ").join(sql.Identifier(column) for column in returning),
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
),
|
|
(record_uuid,),
|
|
)
|
|
before = _row_json(await cur.fetchone())
|
|
if not before:
|
|
raise HTTPException(404, "记录不存在")
|
|
assignments = [
|
|
sql.SQL("{}=%s").format(sql.Identifier(column))
|
|
for column in payload
|
|
]
|
|
if any(field.code == "version" for field in table.fields):
|
|
assignments.append(sql.SQL("version=version+1"))
|
|
assignments.append(sql.SQL("updated_at=now()"))
|
|
await cur.execute(
|
|
sql.SQL("UPDATE {}.{} SET {} WHERE id=%s RETURNING {}").format(
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
sql.SQL(", ").join(assignments),
|
|
sql.SQL(", ").join(sql.Identifier(column) for column in returning),
|
|
),
|
|
[*payload.values(), record_uuid],
|
|
)
|
|
after = _row_json(await cur.fetchone())
|
|
await _write_audit(
|
|
cur,
|
|
schema_name=schema_name,
|
|
tenant_id=tenant_id,
|
|
project_id=project_id,
|
|
table_code=table_code,
|
|
record_id=record_uuid,
|
|
operation="update",
|
|
before_data=before,
|
|
after_data=after,
|
|
actor=actor,
|
|
)
|
|
await conn.commit()
|
|
return after or {}
|
|
|
|
|
|
async def delete_record(
|
|
project_id: str,
|
|
table_code: str,
|
|
record_id: str,
|
|
actor: str,
|
|
) -> dict[str, bool]:
|
|
table = _table_or_404(table_code)
|
|
if not table.allow_delete:
|
|
raise HTTPException(403, "该表不允许删除")
|
|
try:
|
|
record_uuid = uuid.UUID(record_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, "记录 ID 格式错误") from exc
|
|
database = await _database_or_404(project_id)
|
|
schema_name = str(database["schema_name"])
|
|
tenant_id = str(database["tenant_id"])
|
|
columns = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
|
|
async with get_conn() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
sql.SQL("SELECT {} FROM {}.{} WHERE id=%s AND deleted_at IS NULL").format(
|
|
sql.SQL(", ").join(sql.Identifier(column) for column in columns),
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
),
|
|
(record_uuid,),
|
|
)
|
|
before = _row_json(await cur.fetchone())
|
|
if not before:
|
|
raise HTTPException(404, "记录不存在")
|
|
await cur.execute(
|
|
sql.SQL(
|
|
"UPDATE {}.{} SET deleted_at=now(), deleted_by=%s, updated_at=now() WHERE id=%s"
|
|
).format(
|
|
sql.Identifier(schema_name),
|
|
sql.Identifier(table.code),
|
|
),
|
|
(actor, record_uuid),
|
|
)
|
|
await _write_audit(
|
|
cur,
|
|
schema_name=schema_name,
|
|
tenant_id=tenant_id,
|
|
project_id=project_id,
|
|
table_code=table_code,
|
|
record_id=record_uuid,
|
|
operation="delete",
|
|
before_data=before,
|
|
after_data=None,
|
|
actor=actor,
|
|
)
|
|
await conn.commit()
|
|
return {"ok": True}
|
|
|