from dataclasses import dataclass, field from typing import Any @dataclass class CompletedCommand: stdout: str = "" stderr: str = "" returncode: int = 0 @dataclass class RetryPolicy: attempts: int = 3 delay: float = 0.0 def __post_init__(self): if self.attempts < 1: raise ValueError("retry attempts must be at least one") @dataclass(frozen=True) class ResourceRef: kind: str number: int def __post_init__(self): if self.kind not in {"issue", "pr"}: raise ValueError("resource kind must be issue or pr") if self.number < 1: raise ValueError("resource number must be positive") @dataclass class Envelope: ok: bool provider: str | None operation: str result: Any = None details: dict[str, Any] = field(default_factory=dict) error: dict[str, Any] | None = None def to_dict(self): value = { "ok": self.ok, "provider": self.provider, "operation": self.operation, } if self.ok: value["result"] = self.result if self.details: value["details"] = self.details else: value["error"] = self.error return value def normalize_label(value): if isinstance(value, str): return {"name": value} if not isinstance(value, dict): return {"name": str(value)} return { "name": value.get("name", value.get("title", "")), **{key: value[key] for key in ("color", "description", "id") if key in value}, } def normalize_resource(value, *, kind, provider): """Normalize the common fields while retaining provider-specific raw data.""" if not isinstance(value, dict): return {"kind": kind, "number": value, "details": {"raw": value}} number = value.get("number", value.get("iid", value.get("id"))) labels = value.get("labels", value.get("label", [])) or [] assignees = value.get("assignees", value.get("assignee", [])) or [] state = str(value.get("state", "")).lower() state = {"opened": "open", "open": "open", "closed": "closed"}.get(state, state) normalized = { "kind": kind, "number": number, "title": value.get("title", ""), "body": value.get("body", value.get("description", "")) or "", "state": state, "labels": [normalize_label(label) for label in labels], "assignees": assignees if isinstance(assignees, list) else [assignees], "author": value.get("author", value.get("author_name", value.get("user"))), "author_association": value.get("author_association", value.get("authorAssociation")), "url": value.get("url", value.get("web_url", value.get("html_url"))), "comments": value.get("comments", value.get("notes", [])) or [], } for key in ("draft", "merged", "createdAt", "updatedAt", "source_branch", "target_branch", "authorAssociation", "author_association", "membership"): if key in value: normalized[key] = value[key] normalized["details"] = {"provider": provider, "raw": value} return normalized