94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
子任务模型模块
|
||
包含子任务和子任务版本相关的数据模型
|
||
"""
|
||
|
||
from sqlalchemy import Column, Integer, String, Text, Boolean, ForeignKey, JSON
|
||
from sqlalchemy.orm import relationship
|
||
from data.models.base import BaseModel
|
||
|
||
class SubTask(BaseModel):
|
||
"""
|
||
子任务模型
|
||
表示一个可重用的子任务
|
||
"""
|
||
__tablename__ = 'subtasks'
|
||
|
||
name = Column(String(100), nullable=False, comment='子任务名称')
|
||
code = Column(String(50), nullable=False, unique=True, comment='子任务编码')
|
||
description = Column(String(500), nullable=True, comment='子任务描述')
|
||
is_system = Column(Boolean, default=False, comment='是否为系统子任务')
|
||
is_enabled = Column(Boolean, default=True, comment='是否启用')
|
||
current_version_id = Column(Integer, nullable=True, comment='当前版本ID')
|
||
|
||
# 关联关系
|
||
versions = relationship('SubTaskVersion', back_populates='subtask', cascade='all, delete-orphan')
|
||
|
||
def __repr__(self):
|
||
return f"<SubTask(id={self.id}, name='{self.name}', code='{self.code}')>"
|
||
|
||
@classmethod
|
||
def get_by_code(cls, code):
|
||
"""
|
||
根据编码获取子任务
|
||
"""
|
||
return cls.query.filter(cls.code == code, cls.is_deleted == False).first()
|
||
|
||
@classmethod
|
||
def get_enabled_subtasks(cls):
|
||
"""
|
||
获取所有启用的子任务
|
||
"""
|
||
return cls.query.filter(cls.is_enabled == True, cls.is_deleted == False).all()
|
||
|
||
@classmethod
|
||
def get_system_subtasks(cls):
|
||
"""
|
||
获取所有系统子任务
|
||
"""
|
||
return cls.query.filter(cls.is_system == True, cls.is_enabled == True, cls.is_deleted == False).all()
|
||
|
||
class SubTaskVersion(BaseModel):
|
||
"""
|
||
子任务版本模型
|
||
表示子任务的一个版本
|
||
"""
|
||
__tablename__ = 'subtask_versions'
|
||
|
||
subtask_id = Column(Integer, ForeignKey('subtasks.id'), nullable=False, comment='子任务ID')
|
||
version = Column(String(20), nullable=False, comment='版本号')
|
||
content = Column(JSON, nullable=True, comment='子任务内容（JSON格式）')
|
||
flow_data = Column(JSON, nullable=True, comment='流程数据（JSON格式）')
|
||
input_schema = Column(JSON, nullable=True, comment='输入参数模式（JSON Schema）')
|
||
output_schema = Column(JSON, nullable=True, comment='输出参数模式（JSON Schema）')
|
||
remark = Column(String(500), nullable=True, comment='版本备注')
|
||
|
||
# 关联关系
|
||
subtask = relationship('SubTask', back_populates='versions')
|
||
|
||
def __repr__(self):
|
||
return f"<SubTaskVersion(id={self.id}, subtask_id={self.subtask_id}, version='{self.version}')>"
|
||
|
||
@classmethod
|
||
def get_latest_version(cls, subtask_id):
|
||
"""
|
||
获取子任务的最新版本
|
||
"""
|
||
return cls.query.filter(
|
||
cls.subtask_id == subtask_id,
|
||
cls.is_deleted == False
|
||
).order_by(cls.created_at.desc()).first()
|
||
|
||
@classmethod
|
||
def get_by_version(cls, subtask_id, version):
|
||
"""
|
||
根据版本号获取子任务版本
|
||
"""
|
||
return cls.query.filter(
|
||
cls.subtask_id == subtask_id,
|
||
cls.version == version,
|
||
cls.is_deleted == False
|
||
).first() |