59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Optional, TYPE_CHECKING
|
|
|
|
from sqlalchemy import String, DateTime, ForeignKey
|
|
from sqlalchemy.dialects.postgresql import BYTEA
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from models.base import BaseModel
|
|
from models.mixins import IdMixin
|
|
|
|
if TYPE_CHECKING:
|
|
from models.auth import User
|
|
|
|
|
|
class CeleryTask(BaseModel, IdMixin):
|
|
__tablename__ = "celery_taskmeta"
|
|
|
|
task_id: Mapped[int] = mapped_column(String(155), unique=True)
|
|
status: Mapped[int] = mapped_column(String(50))
|
|
result: Mapped[bytes] = mapped_column(BYTEA)
|
|
date_done: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=lambda: datetime.now(timezone.utc),
|
|
)
|
|
traceback: Mapped[str] = mapped_column()
|
|
name: Mapped[str] = mapped_column(String(155))
|
|
args: Mapped[bytes] = mapped_column(BYTEA)
|
|
kwargs: Mapped[bytes] = mapped_column(BYTEA)
|
|
worker: Mapped[str] = mapped_column(String(155))
|
|
retries: Mapped[int] = mapped_column()
|
|
queue: Mapped[str] = mapped_column(String(155))
|
|
|
|
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
|
|
user: Mapped["User"] = relationship(
|
|
back_populates="celery_tasks",
|
|
lazy="noload",
|
|
)
|
|
|
|
|
|
class CeleryActiveTask(BaseModel):
|
|
__tablename__ = "celery_taskmeta_active"
|
|
|
|
task_id: Mapped[str] = mapped_column(
|
|
String(155),
|
|
primary_key=True,
|
|
)
|
|
status: Mapped[str] = mapped_column(String(50))
|
|
name: Mapped[str] = mapped_column(String(155))
|
|
is_notification_hidden: Mapped[bool] = mapped_column(
|
|
default=False,
|
|
server_default="0",
|
|
)
|
|
|
|
user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
|
|
user: Mapped[Optional["User"]] = relationship(
|
|
back_populates="celery_active_tasks",
|
|
lazy="noload",
|
|
)
|