122 lines
2.5 KiB
Python
122 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class TaskState(StrEnum):
|
|
DISCOVERED = "DISCOVERED"
|
|
CLAIMED = "CLAIMED"
|
|
PLANNING = "PLANNING"
|
|
IMPLEMENTING = "IMPLEMENTING"
|
|
TESTING = "TESTING"
|
|
PR_OPENED = "PR_OPENED"
|
|
REVIEWING = "REVIEWING"
|
|
HUMAN_REVIEW_READY = "HUMAN_REVIEW_READY"
|
|
BLOCKED = "BLOCKED"
|
|
FAILED = "FAILED"
|
|
CANCELLED = "CANCELLED"
|
|
|
|
|
|
ACTIVE_STATES = {
|
|
TaskState.DISCOVERED,
|
|
TaskState.CLAIMED,
|
|
TaskState.PLANNING,
|
|
TaskState.IMPLEMENTING,
|
|
TaskState.TESTING,
|
|
TaskState.PR_OPENED,
|
|
TaskState.REVIEWING,
|
|
}
|
|
|
|
TERMINAL_STATES = {
|
|
TaskState.HUMAN_REVIEW_READY,
|
|
TaskState.BLOCKED,
|
|
TaskState.FAILED,
|
|
TaskState.CANCELLED,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RepositoryRecord:
|
|
id: int
|
|
owner: str
|
|
name: str
|
|
full_name: str
|
|
clone_url: str
|
|
default_branch: str
|
|
enabled: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IssueRecord:
|
|
id: int
|
|
repo_id: int
|
|
issue_number: int
|
|
title: str
|
|
body: str
|
|
labels: list[str]
|
|
state: str
|
|
html_url: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TaskRecord:
|
|
id: int
|
|
repo_id: int
|
|
issue_number: int
|
|
state: TaskState
|
|
lease_owner: str | None
|
|
lease_expires_at: datetime | None
|
|
branch_name: str | None
|
|
workspace_path: Path | None
|
|
pr_number: int | None
|
|
error_message: str | None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentResult:
|
|
exit_code: int
|
|
stdout: str
|
|
stderr: str
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.exit_code == 0
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc).replace(microsecond=0)
|
|
|
|
|
|
def parse_dt(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
if value.endswith("Z"):
|
|
value = value[:-1] + "+00:00"
|
|
return datetime.fromisoformat(value)
|
|
|
|
|
|
def dt_to_db(value: datetime | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc).isoformat()
|
|
|
|
|
|
def labels_from_gitea(labels: list[dict[str, Any]] | list[str] | None) -> list[str]:
|
|
names: list[str] = []
|
|
for label in labels or []:
|
|
if isinstance(label, str):
|
|
names.append(label)
|
|
else:
|
|
name = label.get("name")
|
|
if name:
|
|
names.append(str(name))
|
|
return names
|