Underground-Digital/Workflow-Engine

1from datetime import datetime, timezone2 3from celery import states4 5from extensions.ext_database import db6 7 8class CeleryTask(db.Model):9    """Task result/status."""10 11    __tablename__ = "celery_taskmeta"12 13    id = db.Column(db.Integer, db.Sequence("task_id_sequence"), primary_key=True, autoincrement=True)14    task_id = db.Column(db.String(155), unique=True)15    status = db.Column(db.String(50), default=states.PENDING)16    result = db.Column(db.PickleType, nullable=True)17    date_done = db.Column(18        db.DateTime,19        default=lambda: datetime.now(timezone.utc).replace(tzinfo=None),20        onupdate=lambda: datetime.now(timezone.utc).replace(tzinfo=None),21        nullable=True,22    )23    traceback = db.Column(db.Text, nullable=True)24    name = db.Column(db.String(155), nullable=True)25    args = db.Column(db.LargeBinary, nullable=True)26    kwargs = db.Column(db.LargeBinary, nullable=True)27    worker = db.Column(db.String(155), nullable=True)28    retries = db.Column(db.Integer, nullable=True)29    queue = db.Column(db.String(155), nullable=True)30 31 32class CeleryTaskSet(db.Model):33    """TaskSet result."""34 35    __tablename__ = "celery_tasksetmeta"36 37    id = db.Column(db.Integer, db.Sequence("taskset_id_sequence"), autoincrement=True, primary_key=True)38    taskset_id = db.Column(db.String(155), unique=True)39    result = db.Column(db.PickleType, nullable=True)40    date_done = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc).replace(tzinfo=None), nullable=True)41