"""审批流程模型 — 支持多级、并行、条件审批、版本控制与执行日志。""" import json from sqlalchemy import ( Column, Integer, String, Text, Boolean, DateTime, ForeignKey, JSON, func, UniqueConstraint ) from sqlalchemy.orm import relationship from app.core.database import Base class WorkflowDefinition(Base): """审批流程定义模板。""" __tablename__ = "workflow_definitions" id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(100), nullable=False) description = Column(Text, nullable=True) category = Column(String(50), default="ticket") # ticket / query / instance / general is_default = Column(Boolean, default=False) is_active = Column(Boolean, default=True) created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) created_by = Column(Integer, ForeignKey("users.id"), nullable=True) # ═══════════════════════════════════════════════════════════ # 流程级配置（JSON格式，灵活扩展） # ═══════════════════════════════════════════════════════════ # 全局流程配置: # { # "timeout_strategy": "auto_approve|auto_reject|escalate|notify", # "timeout_hours": 48, # "escalation_user_id": 1, # "allow_transfer": true, # "allow_skip": false, # "require_comment": true, # "parallel_mode": "all|any", // 并行审批模式：全部通过/任一通过 # "notify_channels": ["email", "sms", "webhook"] # } flow_config = Column(JSON, nullable=True, default=dict) # 触发条件配置（支持多条件组合）: # { # "conditions": [ # {"field": "ticket.type", "operator": "in", "value": ["DDL", "DML"]}, # {"field": "ticket.risk_level", "operator": ">=", "value": "high"}, # {"field": "database.instance_id", "operator": "==", "value": 1} # ], # "logic": "AND|OR" # } trigger_conditions = Column(JSON, nullable=True) # 步骤定义关联 steps = relationship( "WorkflowStep", back_populates="workflow", order_by="WorkflowStep.step_number", cascade="all, delete-orphan" ) # 版本关联 versions = relationship( "WorkflowVersion", back_populates="workflow", order_by="desc(WorkflowVersion.version_number)", cascade="all, delete-orphan" ) # 审批实例关联 approvals = relationship( "TicketApproval", back_populates="workflow_def", lazy="dynamic" ) # 执行日志关联 execution_logs = relationship( "WorkflowExecutionLog", back_populates="workflow", lazy="dynamic", order_by="desc(WorkflowExecutionLog.created_at)" ) def get_current_version(self): """获取当前激活的版本。""" return next((v for v in self.versions if v.is_active), None) def to_dict(self, include_steps: bool = False) -> dict: data = { "id": self.id, "name": self.name, "description": self.description, "category": self.category, "is_default": self.is_default, "is_active": self.is_active, "flow_config": self.flow_config, "trigger_conditions": self.trigger_conditions, "step_count": len(self.steps) if self.steps else 0, "version_count": len(self.versions) if self.versions else 0, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } if include_steps: data["steps"] = [s.to_dict() for s in (self.steps or [])] return data class WorkflowVersion(Base): """审批流程版本控制 — 支持流程的版本管理与回溯。""" __tablename__ = "workflow_versions" __table_args__ = (UniqueConstraint("workflow_id", "version_number", name="uix_workflow_version"),) id = Column(Integer, primary_key=True, autoincrement=True) workflow_id = Column(Integer, ForeignKey("workflow_definitions.id", ondelete="CASCADE"), nullable=False) version_number = Column(Integer, nullable=False) # 版本号，递增 version_name = Column(String(100), nullable=True) # 版本名称，如 "v2.0-添加并行审批" is_active = Column(Boolean, default=False) # 是否为当前激活版本 # 版本快照 — 存储此版本完整的流程定义快照 steps_snapshot = Column(JSON, nullable=False) flow_config_snapshot = Column(JSON, nullable=True) # 变更说明 change_summary = Column(Text, nullable=True) created_by = Column(Integer, ForeignKey("users.id"), nullable=True) created_at = Column(DateTime, server_default=func.now()) # 关联 workflow = relationship("WorkflowDefinition", back_populates="versions") def to_dict(self) -> dict: return { "id": self.id, "workflow_id": self.workflow_id, "version_number": self.version_number, "version_name": self.version_name, "is_active": self.is_active, "change_summary": self.change_summary, "steps_count": len(self.steps_snapshot) if self.steps_snapshot else 0, "created_at": self.created_at.isoformat() if self.created_at else None, } class WorkflowStep(Base): """审批步骤定义 — 支持灵活的条件判断与动态审批人分配。""" __tablename__ = "workflow_steps" id = Column(Integer, primary_key=True, autoincrement=True) workflow_id = Column(Integer, ForeignKey("workflow_definitions.id", ondelete="CASCADE"), nullable=False) step_number = Column(Integer, nullable=False) name = Column(String(100), nullable=False) description = Column(Text, nullable=True) # 步骤类型: serial(串行) / parallel(并行-会签) / any_one(或签) / condition(条件分支) step_type = Column(String(20), nullable=False, default="serial") # 审批人角色 approver_role = Column(String(20), nullable=True) # admin / dba / manager / creator # 指定审批人ID列表（JSON数组） approver_ids = Column(JSON, nullable=True) # 审批人分配策略: # { # "type": "role|specific|leader|load_balance|creator|expression", # "expression": "${ticket.creator.manager_id}", // 动态表达式 # "fallback_user_id": 1, // 无法解析时的 fallback # "department_id": 2 // 指定部门 # } approver_assignment = Column(JSON, nullable=True, default=dict) is_required = Column(Boolean, default=True) timeout_hours = Column(Integer, default=24) auto_approve_on_timeout = Column(Boolean, default=False) # ═══════════════════════════════════════════════════════════ # 条件判断配置（条件步骤使用） # ═══════════════════════════════════════════════════════════ # 步骤准入条件（满足条件才进入此步骤）: # { # "conditions": [ # {"field": "ticket.sql_type", "operator": "in", "value": ["ALTER", "DROP"]}, # {"field": "affected_rows", "operator": ">", "value": 1000} # ], # "logic": "AND" # } entry_conditions = Column(JSON, nullable=True) # 步骤流转条件（满足条件才进入下一步，否则跳过后续步骤）: # { # "conditions": [ # {"field": "current.result", "operator": "==", "value": "approved"} # ], # "on_false_action": "skip|reject|goto_step", # "goto_step_number": 3 # } transition_conditions = Column(JSON, nullable=True) # ═══════════════════════════════════════════════════════════ # 字段权限配置 # ═══════════════════════════════════════════════════════════ # 此步骤中可查看/编辑的字段: # { # "visible_fields": ["ticket.title", "ticket.sql", "ticket.description"], # "editable_fields": ["ticket.comment"], # "masked_fields": ["ticket.password", "ticket.secret_key"] # } field_permissions = Column(JSON, nullable=True) workflow = relationship("WorkflowDefinition", back_populates="steps") def to_dict(self) -> dict: return { "id": self.id, "workflow_id": self.workflow_id, "step_number": self.step_number, "name": self.name, "description": self.description, "step_type": self.step_type, "approver_role": self.approver_role, "approver_ids": self.approver_ids, "approver_assignment": self.approver_assignment, "is_required": self.is_required, "timeout_hours": self.timeout_hours, "auto_approve_on_timeout": self.auto_approve_on_timeout, "entry_conditions": self.entry_conditions, "transition_conditions": self.transition_conditions, "field_permissions": self.field_permissions, } class TicketApproval(Base): """工单审批实例（运行时记录）。""" __tablename__ = "ticket_approvals" id = Column(Integer, primary_key=True, autoincrement=True) ticket_id = Column(Integer, ForeignKey("tickets.id"), nullable=False, index=True) # 关联的流程定义和版本 workflow_id = Column(Integer, ForeignKey("workflow_definitions.id"), nullable=True) version_id = Column(Integer, ForeignKey("workflow_versions.id"), nullable=True) step_number = Column(Integer, nullable=False) step_name = Column(String(100), nullable=True) step_type = Column(String(20), default="serial") # serial / parallel / any_one / condition # 审批人信息 approver_id = Column(Integer, ForeignKey("users.id"), nullable=True) approver_role = Column(String(20), nullable=True) # 实际执行人（转交场景） actual_approver_id = Column(Integer, ForeignKey("users.id"), nullable=True) # 审批动作: pending / approved / rejected / transferred / skipped / auto_approved action = Column(String(20), default="pending", index=True) comment = Column(Text, nullable=True) # 转交信息 transfer_from = Column(Integer, ForeignKey("users.id"), nullable=True) transfer_reason = Column(Text, nullable=True) # 审批时间记录 created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) completed_at = Column(DateTime, nullable=True) # 超时标记 is_timeout = Column(Boolean, default=False) timeout_handled_at = Column(DateTime, nullable=True) # 关联 workflow_def = relationship("WorkflowDefinition", back_populates="approvals") approver = relationship("User", foreign_keys=[approver_id]) actual_approver = relationship("User", foreign_keys=[actual_approver_id]) def to_dict(self) -> dict: return { "id": self.id, "ticket_id": self.ticket_id, "workflow_id": self.workflow_id, "step_number": self.step_number, "step_name": self.step_name, "step_type": self.step_type, "approver_id": self.approver_id, "approver_role": self.approver_role, "approver_name": self.approver.display_name if self.approver else None, "actual_approver_id": self.actual_approver_id, "action": self.action, "comment": self.comment, "transfer_from": self.transfer_from, "transfer_reason": self.transfer_reason, "is_timeout": self.is_timeout, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "completed_at": self.completed_at.isoformat() if self.completed_at else None, } class WorkflowExecutionLog(Base): """流程执行日志 — 记录审批流程的完整执行轨迹，用于审计和问题排查。""" __tablename__ = "workflow_execution_logs" id = Column(Integer, primary_key=True, autoincrement=True) workflow_id = Column(Integer, ForeignKey("workflow_definitions.id"), nullable=False, index=True) ticket_id = Column(Integer, ForeignKey("tickets.id"), nullable=True, index=True) step_id = Column(Integer, ForeignKey("workflow_steps.id"), nullable=True) approval_id = Column(Integer, ForeignKey("ticket_approvals.id"), nullable=True) # 事件类型: # workflow_started / step_entered / step_completed / approved / rejected / # transferred / skipped / timeout / escalated / workflow_completed / workflow_cancelled event_type = Column(String(50), nullable=False, index=True) # 事件详情 event_description = Column(Text, nullable=True) # 执行人 actor_id = Column(Integer, ForeignKey("users.id"), nullable=True) actor_name = Column(String(100), nullable=True) # 上下文数据（快照）: # { # "step_snapshot": {...}, # "approval_snapshot": {...}, # "decision_context": {...}, # "condition_results": [...] # } context_data = Column(JSON, nullable=True) created_at = Column(DateTime, server_default=func.now(), index=True) # 关联 workflow = relationship("WorkflowDefinition", back_populates="execution_logs") def to_dict(self) -> dict: return { "id": self.id, "workflow_id": self.workflow_id, "ticket_id": self.ticket_id, "step_id": self.step_id, "approval_id": self.approval_id, "event_type": self.event_type, "event_description": self.event_description, "actor_id": self.actor_id, "actor_name": self.actor_name, "context_data": self.context_data, "created_at": self.created_at.isoformat() if self.created_at else None, } def __repr__(self) -> str: return ( f"" )