""" IssueSink adapter interface and implementations. IssueSink is the outbound boundary between activity-core and task backends (issue-core, etc.). It receives TaskSpec objects and returns TaskRef objects. Active sink is selected by ISSUE_SINK_TYPE env var: "rest" (default) | "null". """ from __future__ import annotations import logging import os import uuid from abc import ABC, abstractmethod import httpx from activity_core.rules.models import TaskRef, TaskSpec logger = logging.getLogger(__name__) ISSUE_CORE_URL = os.environ.get("ISSUE_CORE_URL", "http://127.0.0.1:8765") ISSUE_CORE_API_KEY_ENV = "ISSUE_CORE_API_KEY" ISSUE_SINK_TYPE = os.environ.get("ISSUE_SINK_TYPE", "rest") class IssueSink(ABC): @abstractmethod def emit(self, task_spec: TaskSpec) -> TaskRef: ... class IssueCoreRestSink(IssueSink): """POSTs to issue-core REST API. Config: ISSUE_CORE_URL and ISSUE_CORE_API_KEY env vars (shared key with the issue-core server). """ def __init__( self, base_url: str = ISSUE_CORE_URL, api_key: str | None = None, ) -> None: self._base_url = base_url.rstrip("/") if api_key is not None: self._api_key = api_key.strip() else: self._api_key = os.environ.get(ISSUE_CORE_API_KEY_ENV, "").strip() def _auth_headers(self) -> dict[str, str]: if not self._api_key: raise RuntimeError( f"{ISSUE_CORE_API_KEY_ENV} is not set. " "Required when ISSUE_SINK_TYPE=rest." ) return {"Authorization": f"Bearer {self._api_key}"} def emit(self, task_spec: TaskSpec) -> TaskRef: payload = { "title": task_spec.title, "description": task_spec.description, "target_repo": task_spec.target_repo, "priority": task_spec.priority, "labels": task_spec.labels, "due_in_days": task_spec.due_in_days, "source_type": task_spec.source_type, "source_id": task_spec.source_id, "triggering_event_id": ( str(task_spec.triggering_event_id) if task_spec.triggering_event_id is not None else None ), "activity_definition_id": task_spec.activity_definition_id, } resp = httpx.post( f"{self._base_url}/issues/", json=payload, headers=self._auth_headers(), timeout=10.0, ) resp.raise_for_status() data = resp.json() return TaskRef( external_id=data["issue_id"], backend_url=data.get("issue_url"), backend=data.get("backend", ""), ) class NullSink(IssueSink): """Discards tasks and returns synthetic TaskRefs. For testing.""" def emit(self, task_spec: TaskSpec) -> TaskRef: synthetic_id = f"null-{uuid.uuid4()}" logger.debug("NullSink: discarding task %r → %s", task_spec.title, synthetic_id) return TaskRef(external_id=synthetic_id, backend="null") def get_issue_sink() -> IssueSink: """Factory: returns the configured IssueSink based on ISSUE_SINK_TYPE.""" sink_type = ISSUE_SINK_TYPE.lower() if sink_type == "null": return NullSink() return IssueCoreRestSink()