ElebobaandClaude Opus 5.5
25735fee96
- SourceCraftClient (httpx): list_runs, get_run, get_cube_logs (постранично), run_commit_sha; ретраи на сетевые ошибки и 5xx с экспоненциальной задержкой, 4xx — SourceCraftError без повторов - модели: RunStatus/EventType из swagger, «1970-01-01» = не задано, failed_cubes без служебных #-кубов - у прогона в API нет sha → куб check/meta в tgit-uploads печатает TGIT_COMMIT_SHA (ADR-014) - CLI: python -m app.sourcecraft.cli runs - тесты на реальных ответах API (фикстуры), порог покрытия 45 → 80 Refs #10 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
114 lines
2.8 KiB
Python
114 lines
2.8 KiB
Python
"""Typed SourceCraft CI/CD API responses (only the fields the bot uses)."""
|
|
|
|
from collections.abc import Iterator
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
|
|
class RunStatus(StrEnum):
|
|
CREATED = "created"
|
|
PREPARED = "prepared"
|
|
PROCESSING = "processing"
|
|
SUCCESS = "success"
|
|
FAILED = "failed"
|
|
CANCELED = "canceled"
|
|
TIMEOUT = "timeout"
|
|
SKIPPED = "skipped"
|
|
AWAITING_APPROVAL = "awaiting_approval"
|
|
REJECTED = "rejected"
|
|
|
|
@property
|
|
def finished(self) -> bool:
|
|
return self in _FINISHED
|
|
|
|
|
|
_FINISHED = {
|
|
RunStatus.SUCCESS,
|
|
RunStatus.FAILED,
|
|
RunStatus.CANCELED,
|
|
RunStatus.TIMEOUT,
|
|
RunStatus.SKIPPED,
|
|
RunStatus.REJECTED,
|
|
}
|
|
|
|
|
|
class EventType(StrEnum):
|
|
PUSH = "push"
|
|
PR_UPDATE = "pr_update"
|
|
MANUAL = "manual"
|
|
RESTART = "restart"
|
|
SCHEDULE = "schedule"
|
|
REPOSITORY_EVENT = "repository_event"
|
|
|
|
|
|
class _Model(BaseModel):
|
|
model_config = ConfigDict(extra="ignore", frozen=True)
|
|
|
|
|
|
class Dates(_Model):
|
|
created_at: datetime | None = None
|
|
started_at: datetime | None = None
|
|
finished_at: datetime | None = None
|
|
|
|
@field_validator("created_at", "started_at", "finished_at", mode="after")
|
|
@classmethod
|
|
def _epoch_is_unset(cls, value: datetime | None) -> datetime | None:
|
|
# The API reports "not yet" as 1970-01-01T00:00:00Z
|
|
return None if value is not None and value.year == 1970 else value
|
|
|
|
|
|
class Cube(_Model):
|
|
slug: str
|
|
status: RunStatus
|
|
dates: Dates = Field(default_factory=Dates)
|
|
|
|
@property
|
|
def is_service(self) -> bool:
|
|
"""Platform steps (#git-clone, #prepare-image-for-…): not in our config, logs not readable."""
|
|
return self.slug.startswith("#")
|
|
|
|
|
|
class Task(_Model):
|
|
slug: str
|
|
status: RunStatus
|
|
cubes: list[Cube] = []
|
|
|
|
|
|
class Workflow(_Model):
|
|
slug: str
|
|
status: RunStatus
|
|
tasks: list[Task] = []
|
|
|
|
|
|
class Run(_Model):
|
|
slug: str
|
|
status: RunStatus
|
|
event_type: EventType
|
|
dates: Dates = Field(default_factory=Dates)
|
|
workflows: list[Workflow] = []
|
|
error_messages: list[str] = []
|
|
|
|
def task(self, workflow: str, task: str) -> Task | None:
|
|
for wf in self.workflows:
|
|
if wf.slug == workflow:
|
|
return next((t for t in wf.tasks if t.slug == task), None)
|
|
return None
|
|
|
|
def cubes(self) -> Iterator[tuple[str, str, Cube]]:
|
|
"""(workflow, task, cube) for every user-defined cube."""
|
|
for wf in self.workflows:
|
|
for t in wf.tasks:
|
|
for c in t.cubes:
|
|
if not c.is_service:
|
|
yield wf.slug, t.slug, c
|
|
|
|
def failed_cubes(self) -> list[tuple[str, str, Cube]]:
|
|
return [entry for entry in self.cubes() if entry[2].status is RunStatus.FAILED]
|
|
|
|
|
|
class CubeLog(_Model):
|
|
text: str
|
|
done: bool
|