JiuContinent
0a153b264a
refactor: 移除审批和回写功能模块 移除了整个审批(approvals)和官方回写(writebacks)功能模块， 包括相关模型、路由、服务和配置项。更新了数据库迁移文件， 删除了相关的审批请求表和官方回写运行表。同时从API路由器中 移除了相应的路由，并调整了安全常量和字段验证器以匹配变更。 ```
119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
from datetime import date, datetime
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import Date as SQLDate
|
|
from sqlalchemy import DateTime as SQLDateTime
|
|
from sqlalchemy import Numeric as SQLNumeric
|
|
from sqlalchemy import Select, func, select
|
|
from sqlalchemy.sql.schema import Column
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.pagination import bounded_limit, bounded_offset
|
|
from app.modules.business.registry import get_domain_model, get_writable_fields
|
|
from app.modules.business.constants import (
|
|
INVALID_FIELD_VALUE_TEMPLATE,
|
|
READ_ONLY_FIELD_TEMPLATE,
|
|
UNKNOWN_FIELD_TEMPLATE,
|
|
BusinessErrorDetail,
|
|
BusinessField,
|
|
)
|
|
|
|
|
|
def serialize_model(record: Any) -> dict[str, Any]:
|
|
"""Convert a SQLAlchemy model instance into a JSON-friendly dictionary."""
|
|
|
|
data: dict[str, Any] = {}
|
|
for column in record.__table__.columns:
|
|
value = getattr(record, column.name)
|
|
if isinstance(value, (datetime, date)):
|
|
data[column.name] = value.isoformat()
|
|
elif isinstance(value, Decimal):
|
|
data[column.name] = float(value)
|
|
else:
|
|
data[column.name] = value
|
|
return data
|
|
|
|
|
|
def _coerce_column_value(column: Column, value: Any) -> Any:
|
|
"""Coerce API JSON values into the Python type expected by a SQLAlchemy column."""
|
|
|
|
if value is None:
|
|
return None
|
|
if isinstance(column.type, SQLDateTime) and isinstance(value, str):
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if isinstance(column.type, SQLDate) and isinstance(value, str):
|
|
return date.fromisoformat(value)
|
|
if isinstance(column.type, SQLNumeric) and not isinstance(value, Decimal):
|
|
return Decimal(str(value))
|
|
return value
|
|
|
|
|
|
def _model_payload(domain: str, model: Any, data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Validate keys and coerce values according to model column types."""
|
|
|
|
columns = {
|
|
column.name: column
|
|
for column in model.__table__.columns
|
|
if column.name != BusinessField.ID
|
|
}
|
|
writable_fields = get_writable_fields(domain)
|
|
payload: dict[str, Any] = {}
|
|
for key, value in data.items():
|
|
column = columns.get(key)
|
|
if column is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=UNKNOWN_FIELD_TEMPLATE.format(field=key),
|
|
)
|
|
if key not in writable_fields:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=READ_ONLY_FIELD_TEMPLATE.format(field=key),
|
|
)
|
|
try:
|
|
payload[key] = _coerce_column_value(column, value)
|
|
except (ValueError, TypeError, InvalidOperation) as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=INVALID_FIELD_VALUE_TEMPLATE.format(field=key),
|
|
) from exc
|
|
return payload
|
|
|
|
|
|
class BusinessService:
|
|
"""Read business records across registered domains."""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def list_records(
|
|
self,
|
|
domain: str,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
status_filter: str | None = None,
|
|
) -> tuple[int, list[dict[str, Any]]]:
|
|
model = get_domain_model(domain)
|
|
stmt: Select = select(model)
|
|
count_stmt = select(func.count()).select_from(model)
|
|
if status_filter and hasattr(model, BusinessField.STATUS):
|
|
stmt = stmt.where(model.status == status_filter)
|
|
count_stmt = count_stmt.where(model.status == status_filter)
|
|
stmt = stmt.order_by(model.id.desc()).limit(bounded_limit(limit)).offset(
|
|
bounded_offset(offset)
|
|
)
|
|
total = int(self.db.execute(count_stmt).scalar() or 0)
|
|
return total, [serialize_model(item) for item in self.db.execute(stmt).scalars()]
|
|
|
|
def get_record(self, domain: str, record_id: int) -> dict[str, Any]:
|
|
model = get_domain_model(domain)
|
|
record = self.db.get(model, record_id)
|
|
if record is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=BusinessErrorDetail.RECORD_NOT_FOUND,
|
|
)
|
|
return serialize_model(record)
|