kyleandClaude Opus 5.5
bf752a904f
- FastAPI + SQLite backend: tenants, jobs and runs, with an idempotent /api/v1/ingest endpoint authenticated by per-collector API keys - Runs store started_at and finished_at; "Last Run" is the most recent finish, and in-progress runs show as Running - Job modal shows a 30-day daily strip, success rate, and a filterable run list with per-run messages - nginx: HTTPS plus basic auth, ingest endpoint exempt; systemd unit - manage.py for API keys, job cleanup and demo data; pytest suite Co-Authored-By: Claude Opus 5.5 <[email protected]>
64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
import json
|
|
from typing import Annotated, Any, Literal
|
|
|
|
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
|
|
|
|
RunStatus = Literal["running", "success", "warning", "failed"]
|
|
Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)]
|
|
|
|
MAX_MESSAGE_CHARS = 32_000
|
|
MAX_DETAILS_BYTES = 16_000
|
|
|
|
|
|
class Strict(BaseModel):
|
|
# Reject unknown fields so a typo in a collector script fails loudly instead of being dropped.
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class JobIn(Strict):
|
|
external_id: Name = Field(description="The source system's own stable ID for the job")
|
|
name: Name
|
|
# Omit to leave the stored value untouched; send null to clear it.
|
|
next_run_at: AwareDatetime | None = None
|
|
|
|
|
|
class RunIn(Strict):
|
|
external_id: Name | None = Field(
|
|
None, description="The source's ID for this run. Defaults to started_at, so a 'running' "
|
|
"report and its later completion report merge into one run.")
|
|
status: RunStatus
|
|
started_at: AwareDatetime
|
|
# Omit for a finished run to use the time the report was received.
|
|
finished_at: AwareDatetime | None = None
|
|
message: str = ""
|
|
size_bytes: int | None = Field(None, ge=0)
|
|
details: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
@field_validator("message")
|
|
@classmethod
|
|
def truncate_message(cls, v: str) -> str:
|
|
# Backup logs can be huge; keep the head, which is where the summary usually is.
|
|
return v if len(v) <= MAX_MESSAGE_CHARS else v[:MAX_MESSAGE_CHARS] + "\n… [truncated]"
|
|
|
|
@field_validator("details")
|
|
@classmethod
|
|
def limit_details(cls, v: dict[str, Any]) -> dict[str, Any]:
|
|
if len(json.dumps(v)) > MAX_DETAILS_BYTES:
|
|
raise ValueError(f"details must serialize to at most {MAX_DETAILS_BYTES} bytes")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def check_times(self) -> "RunIn":
|
|
if self.status == "running" and self.finished_at is not None:
|
|
raise ValueError("a running run cannot have finished_at")
|
|
if self.finished_at is not None and self.finished_at < self.started_at:
|
|
raise ValueError("finished_at is before started_at")
|
|
return self
|
|
|
|
|
|
class IngestIn(Strict):
|
|
tenant: Name = Field(description="Company name as shown on the dashboard")
|
|
source_system: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=100)]
|
|
job: JobIn
|
|
run: RunIn | None = Field(None, description="Omit to only register/update the job (e.g. next_run_at)")
|