from dataclasses import dataclass from datetime import datetime from typing import Optional from sqlalchemy import Column, Integer, String, Boolean, Text, DateTime from database.database import Base @dataclass class TScheduler(Base): __tablename__ = 't_scheduler' id: int = Column(Integer, primary_key=True, autoincrement=True, comment='自动递增的唯一任务ID') task_name: str = Column(String(64), nullable=False, comment='任务名称') trigger: str = Column(String(10), nullable=False, comment='调度方式,interval、cron、date') interval_seconds: Optional[int] = Column(Integer, nullable=True, comment='固定时间间隔(秒),用于 interval 类型') cron_expression: Optional[str] = Column(String(255), nullable=True, comment='CRON 表达式,用于 cron 类型') execution_date: Optional[datetime] = Column(DateTime, nullable=True, comment='执行时间,用于 date 类型') task_payload: Optional[str] = Column(Text, nullable=True, comment='任务相关的参数或数据') active: Optional[bool] = Column(Boolean, default=False, nullable=True, comment='任务状态,是否启用') executor: Optional[str] = Column(String(32), nullable=True, comment='任务执行者') description: Optional[str] = Column(String(256), nullable=True, comment='任务描述') last_run: Optional[datetime] = Column(DateTime, nullable=True, comment='上一次执行时间') next_run: Optional[datetime] = Column(DateTime, nullable=True, comment='下一次执行时间') create_time: datetime = Column(DateTime, default=datetime.utcnow, nullable=True, comment='创建时间') update_time: datetime = Column(DateTime, default=datetime.utcnow, nullable=True, comment='更新时间') module_path: Optional[str] = Column(String(255), nullable=True, comment='任务逻辑所在模块名称') function_name: Optional[str] = Column(String(256), nullable=True, comment='任务逻辑的函数名称')