feat: add job models with Pydantic

app/models.py

@@ -0,0 +1,48 @@
+import secrets
+from datetime import datetime, timezone
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class JobStatus(str, Enum):
+    QUEUED = "queued"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class Progress(BaseModel):
+    frame: int = 0
+    fps: float = 0.0
+    time: str = "00:00:00.00"
+    bitrate: str = "0kbits/s"
+    percent: float | None = None
+
+
+class Job(BaseModel):
+    id: str = Field(default_factory=lambda: f"job_{secrets.token_hex(8)}")
+    status: JobStatus = JobStatus.QUEUED
+    command: str
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+    progress: Progress | None = None
+    output_files: list[str] = Field(default_factory=list)
+    error: str | None = None
+
+
+class CreateJobRequest(BaseModel):
+    command: str
+
+
+class JobResponse(BaseModel):
+    id: str
+    status: JobStatus
+    command: str
+    created_at: datetime
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+    progress: Progress | None = None
+    output_files: list[str] = Field(default_factory=list)
+    error: str | None = None

requirements.txt

@@ -1,3 +1,3 @@
-fastapi==0.115.0
+fastapi==0.123.0
-uvicorn[standard]==0.32.0
+uvicorn[standard]==0.38.0
-pydantic==2.9.0
+pydantic==2.12.5

tests/test_models.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+
+from app.models import Job, JobStatus, CreateJobRequest, JobResponse
+
+
+def test_job_creation():
+    job = Job(command="-i input.mp4 output.mp4")
+
+    assert job.id.startswith("job_")
+    assert len(job.id) == 20  # job_ + 16 hex chars
+    assert job.status == JobStatus.QUEUED
+    assert job.command == "-i input.mp4 output.mp4"
+    assert isinstance(job.created_at, datetime)
+    assert job.started_at is None
+    assert job.completed_at is None
+    assert job.progress is None
+    assert job.output_files == []
+    assert job.error is None
+
+
+def test_create_job_request():
+    request = CreateJobRequest(command="-i test.mp4 out.mp4")
+    assert request.command == "-i test.mp4 out.mp4"
+
+
+def test_job_response_from_job():
+    job = Job(command="-i input.mp4 output.mp4")
+    response = JobResponse.model_validate(job.model_dump())
+
+    assert response.id == job.id
+    assert response.status == JobStatus.QUEUED