Model/infrastructure/postgres/task_table.py
2026-05-19 16:35:09 +00:00

36 lines
1.1 KiB
Python

from datetime import datetime, timezone
from typing import ClassVar, Optional
from uuid import UUID, uuid4
from sqlalchemy import Column
from sqlalchemy import Enum as SAEnum
from sqlmodel import Field, SQLModel
from domain.tasks.tasks import Source
class TaskRow(SQLModel, table=True):
__tablename__: ClassVar[str] = "tasks" # pyright: ignore[reportIncompatibleVariableOverride]
id: UUID = Field(default_factory=uuid4, primary_key=True, index=True)
task_source: str
job_started: Optional[datetime] = None
job_completed: Optional[datetime] = None
status: str = Field(default="waiting")
service: Optional[str] = None
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
source: Optional[Source] = Field(
default=None,
sa_column=Column(
SAEnum(
Source,
name="source",
values_callable=lambda cls: [m.value for m in cls], # pyright: ignore[reportUnknownLambdaType, reportUnknownMemberType, reportUnknownVariableType]
),
nullable=True,
),
)
source_id: Optional[str] = None