JiuContinent
19e59e83cc
feat: 添加数据库迁移脚本并更新Dockerfile配置 - 在Dockerfile中添加alembic配置文件和目录的复制指令 - 更新alembic/env.py注册新的模块模型：events、workflows、writebacks - 生成完整的初始数据库schema迁移脚本，包含以下表： - approval_requests, attendance_records, audit_logs, domain_events - expenses, feishu_event_receipts, fund_accounts, legacy_sync_runs - official_writeback_runs, performance_metrics, policies, procurements - projects, report_push_runs, risk_event_actions, risk_events - standards, suppliers, work_reports, work_tasks, workflow_actions - workflow_instances等21个数据表结构定义 - 在API路由器中添加新模块的路由：events、workflows、writebacks、observability ```
135 lines
4.5 KiB
Python
135 lines
4.5 KiB
Python
from typing import Any
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.constants import ActorValue
|
|
from app.core.pagination import bounded_limit
|
|
from app.core.time import utc_now
|
|
from app.modules.business.service import serialize_model
|
|
from app.modules.workflows.constants import (
|
|
WORKFLOW_ACTION_CODE_PREFIX,
|
|
WORKFLOW_CODE_PREFIX,
|
|
WorkflowErrorDetail,
|
|
WorkflowStatus,
|
|
)
|
|
from app.modules.workflows.models import WorkflowAction, WorkflowInstance
|
|
|
|
|
|
class WorkflowService:
|
|
"""Track V3 workflow instances and append-only workflow actions."""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def start_or_update(
|
|
self,
|
|
workflow_type: str,
|
|
aggregate_type: str,
|
|
aggregate_id: str | int | None,
|
|
status_value: str,
|
|
action: str,
|
|
actor: str = ActorValue.SYSTEM,
|
|
payload: dict[str, Any] | None = None,
|
|
) -> WorkflowInstance:
|
|
aggregate_id_text = str(aggregate_id) if aggregate_id is not None else None
|
|
record = self.db.execute(
|
|
select(WorkflowInstance).where(
|
|
WorkflowInstance.workflow_type == workflow_type,
|
|
WorkflowInstance.aggregate_type == aggregate_type,
|
|
WorkflowInstance.aggregate_id == aggregate_id_text,
|
|
)
|
|
).scalar_one_or_none()
|
|
previous_status = None
|
|
now = utc_now()
|
|
if record is None:
|
|
record = WorkflowInstance(
|
|
code=f"{WORKFLOW_CODE_PREFIX}-{now:%Y%m%d%H%M%S%f}",
|
|
workflow_type=workflow_type,
|
|
aggregate_type=aggregate_type,
|
|
aggregate_id=aggregate_id_text,
|
|
status=status_value,
|
|
actor=actor,
|
|
current_step=action,
|
|
payload=payload or {},
|
|
)
|
|
self.db.add(record)
|
|
self.db.flush()
|
|
else:
|
|
previous_status = record.status
|
|
record.status = status_value
|
|
record.actor = actor
|
|
record.current_step = action
|
|
record.payload = payload or {}
|
|
record.updated_at = now
|
|
|
|
if status_value in {
|
|
WorkflowStatus.BLOCKED,
|
|
WorkflowStatus.COMPLETED,
|
|
WorkflowStatus.FAILED,
|
|
}:
|
|
record.completed_at = now
|
|
elif previous_status in {
|
|
WorkflowStatus.BLOCKED,
|
|
WorkflowStatus.COMPLETED,
|
|
WorkflowStatus.FAILED,
|
|
}:
|
|
record.completed_at = None
|
|
|
|
self.db.add(
|
|
WorkflowAction(
|
|
code=f"{WORKFLOW_ACTION_CODE_PREFIX}-{now:%Y%m%d%H%M%S%f}",
|
|
workflow_code=record.code,
|
|
action=action,
|
|
actor=actor,
|
|
from_status=previous_status,
|
|
to_status=status_value,
|
|
payload=payload or {},
|
|
)
|
|
)
|
|
self.db.commit()
|
|
self.db.refresh(record)
|
|
return record
|
|
|
|
def list_workflows(
|
|
self,
|
|
status_filter: str | None = None,
|
|
workflow_type: str | None = None,
|
|
limit: int = 100,
|
|
) -> list[dict[str, Any]]:
|
|
stmt = (
|
|
select(WorkflowInstance)
|
|
.order_by(WorkflowInstance.id.desc())
|
|
.limit(bounded_limit(limit))
|
|
)
|
|
if status_filter:
|
|
stmt = stmt.where(WorkflowInstance.status == status_filter)
|
|
if workflow_type:
|
|
stmt = stmt.where(WorkflowInstance.workflow_type == workflow_type)
|
|
return [serialize_model(item) for item in self.db.execute(stmt).scalars()]
|
|
|
|
def get_workflow(self, code: str) -> dict[str, Any]:
|
|
record = self.db.execute(
|
|
select(WorkflowInstance).where(WorkflowInstance.code == code)
|
|
).scalar_one_or_none()
|
|
if record is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=WorkflowErrorDetail.WORKFLOW_NOT_FOUND,
|
|
)
|
|
actions = self.db.execute(
|
|
select(WorkflowAction)
|
|
.where(WorkflowAction.workflow_code == code)
|
|
.order_by(WorkflowAction.id.asc())
|
|
).scalars()
|
|
data = serialize_model(record)
|
|
data["actions"] = [serialize_model(item) for item in actions]
|
|
return data
|
|
|
|
def count_by_status(self) -> dict[str, int]:
|
|
rows = self.db.execute(
|
|
select(WorkflowInstance.status, func.count()).group_by(WorkflowInstance.status)
|
|
).all()
|
|
return {str(status_value): int(count) for status_value, count in rows}
|