Saltar a contenido

API reference

Generated from the apogee-ai-ops source with mkdocstrings. Every symbol below is exported from apogee_ai_ops, so it is part of the supported public surface.

Application · DTOs

AlertDTO dataclass

Python
AlertDTO(title: str, message: str, severity: str = 'warning', sink: str = 'console')

title instance-attribute

Python
title: str

message instance-attribute

Python
message: str

severity class-attribute instance-attribute

Python
severity: str = 'warning'

sink class-attribute instance-attribute

Python
sink: str = 'console'

BenchDTO dataclass

Python
BenchDTO(deployments: int = 50)

deployments class-attribute instance-attribute

Python
deployments: int = 50

DeployDTO dataclass

Python
DeployDTO(name: str, image: str = '', deployer: str = 'dry_run', replicas: int = 1, env: dict[str, str] = dict())

name instance-attribute

Python
name: str

image class-attribute instance-attribute

Python
image: str = ''

deployer class-attribute instance-attribute

Python
deployer: str = 'dry_run'

replicas class-attribute instance-attribute

Python
replicas: int = 1

env class-attribute instance-attribute

Python
env: dict[str, str] = field(default_factory=dict)

HealthDTO dataclass

Python
HealthDTO(target: str, kind: str = 'http')

target instance-attribute

Python
target: str

kind class-attribute instance-attribute

Python
kind: str = 'http'

Application · Use cases

AlertOnFailUseCase

Python
AlertOnFailUseCase(checker, sink, severity_when_unhealthy: AlertSeverity = ERROR)

Runs a healthcheck and emits an alert if not healthy.

Source code in apogee_ai_ops/application/use_cases/alert_on_fail_use_case.py
Python
def __init__(self, checker, sink, severity_when_unhealthy: AlertSeverity = AlertSeverity.ERROR) -> None:
    self._checker = checker
    self._sink = sink
    self._severity = severity_when_unhealthy

execute async

Python
execute(target: str, source: str = 'ops') -> bool
Source code in apogee_ai_ops/application/use_cases/alert_on_fail_use_case.py
Python
async def execute(self, target: str, source: str = "ops") -> bool:
    health = await self._checker.check(target)
    if health.status is HealthStatus.HEALTHY:
        return True
    await self._sink.emit(Alert(
        title=f"{target} is {health.status.value}",
        message=health.message or "no message",
        severity=self._severity,
        source=source,
        metadata={"latency_ms": f"{health.latency_ms:.1f}"},
    ))
    return False

BenchOpsUseCase

execute async

Python
execute(deployments: int) -> dict[str, float]
Source code in apogee_ai_ops/application/use_cases/bench_ops_use_case.py
Python
async def execute(self, deployments: int) -> dict[str, float]:
    if deployments <= 0:
        raise ValueError("deployments must be positive")
    deployer = DryRunDeployer()
    use_case = DeployUseCase(deployer)
    start = time.perf_counter()
    for i in range(deployments):
        await use_case.execute(Deployment(
            name=f"bench-{i:04d}", image="apogee/echo:0.1",
        ))
    elapsed = (time.perf_counter() - start) * 1000.0
    return {
        "deployments": float(deployments),
        "elapsed_ms": elapsed,
        "deployments_per_second": (deployments / elapsed * 1000.0) if elapsed > 0 else 0.0,
    }

DeployUseCase

Python
DeployUseCase(deployer)
Source code in apogee_ai_ops/application/use_cases/deploy_use_case.py
Python
def __init__(self, deployer) -> None:
    self._deployer = deployer

execute async

Python
execute(deployment: Deployment) -> Deployment
Source code in apogee_ai_ops/application/use_cases/deploy_use_case.py
Python
async def execute(self, deployment: Deployment) -> Deployment:
    return await self._deployer.deploy(deployment)

HealthcheckUseCase

Python
HealthcheckUseCase(checker)
Source code in apogee_ai_ops/application/use_cases/healthcheck_use_case.py
Python
def __init__(self, checker) -> None:
    self._checker = checker

execute async

Python
execute(target: str) -> ServiceHealth
Source code in apogee_ai_ops/application/use_cases/healthcheck_use_case.py
Python
async def execute(self, target: str) -> ServiceHealth:
    return await self._checker.check(target)

RollbackUseCase

Python
RollbackUseCase(deployer)
Source code in apogee_ai_ops/application/use_cases/rollback_use_case.py
Python
def __init__(self, deployer) -> None:
    self._deployer = deployer

execute async

Python
execute(name: str) -> None
Source code in apogee_ai_ops/application/use_cases/rollback_use_case.py
Python
async def execute(self, name: str) -> None:
    await self._deployer.rollback(name)

RunRunbookUseCase

Python
RunRunbookUseCase(step_dispatch: Callable[[str, str], Awaitable[None]])

Executes a Runbook step-by-step with on_failure semantics.

step_dispatch is a callable that receives a step name + action and returns a coroutine. Failures honour step.on_failure: stop, continue, or rollback (which re-runs the step's action prefixed with 'rollback:').

Source code in apogee_ai_ops/application/use_cases/run_runbook_use_case.py
Python
def __init__(self, step_dispatch: Callable[[str, str], Awaitable[None]]) -> None:
    self._dispatch = step_dispatch

execute async

Python
execute(runbook: Runbook) -> dict
Source code in apogee_ai_ops/application/use_cases/run_runbook_use_case.py
Python
async def execute(self, runbook: Runbook) -> dict:
    executed: list[str] = []
    failed: list[str] = []
    for step in runbook.steps:
        try:
            await self._dispatch(step.name, step.action)
            executed.append(step.name)
        except Exception as exc:  # noqa: BLE001
            failed.append(step.name)
            if step.on_failure == "stop":
                return {
                    "executed": executed,
                    "failed": failed,
                    "stopped_at": step.name,
                    "error": str(exc),
                }
            if step.on_failure == "rollback":
                try:
                    await self._dispatch(step.name, f"rollback:{step.action}")
                except Exception:  # noqa: BLE001
                    pass
                return {
                    "executed": executed,
                    "failed": failed,
                    "rolled_back_at": step.name,
                    "error": str(exc),
                }
            # on_failure == "continue"
    return {"executed": executed, "failed": failed}

Domain

Alert dataclass

Python
Alert(title: str, message: str, severity: AlertSeverity = WARNING, source: str = 'apogee-ai-ops', timestamp_s: float = time(), metadata: dict[str, str] = dict())

title instance-attribute

Python
title: str

message instance-attribute

Python
message: str

severity class-attribute instance-attribute

Python
severity: AlertSeverity = WARNING

source class-attribute instance-attribute

Python
source: str = 'apogee-ai-ops'

timestamp_s class-attribute instance-attribute

Python
timestamp_s: float = field(default_factory=time)

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

AlertSeverity

Bases: str, Enum

INFO class-attribute instance-attribute

Python
INFO = 'info'

WARNING class-attribute instance-attribute

Python
WARNING = 'warning'

ERROR class-attribute instance-attribute

Python
ERROR = 'error'

CRITICAL class-attribute instance-attribute

Python
CRITICAL = 'critical'

Deployment dataclass

Python
Deployment(name: str, image: str = '', command: tuple[str, ...] = (), env: dict[str, str] = dict(), replicas: int = 1, status: DeploymentStatus = PENDING, metadata: dict[str, str] = dict())

name instance-attribute

Python
name: str

image class-attribute instance-attribute

Python
image: str = ''

command class-attribute instance-attribute

Python
command: tuple[str, ...] = ()

env class-attribute instance-attribute

Python
env: dict[str, str] = field(default_factory=dict)

replicas class-attribute instance-attribute

Python
replicas: int = 1

status class-attribute instance-attribute

Python
status: DeploymentStatus = PENDING

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

DeploymentStatus

Bases: str, Enum

PENDING class-attribute instance-attribute

Python
PENDING = 'pending'

RUNNING class-attribute instance-attribute

Python
RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute

Python
SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute

Python
FAILED = 'failed'

ROLLED_BACK class-attribute instance-attribute

Python
ROLLED_BACK = 'rolled_back'

HealthStatus

Bases: str, Enum

HEALTHY class-attribute instance-attribute

Python
HEALTHY = 'healthy'

DEGRADED class-attribute instance-attribute

Python
DEGRADED = 'degraded'

UNHEALTHY class-attribute instance-attribute

Python
UNHEALTHY = 'unhealthy'

UNKNOWN class-attribute instance-attribute

Python
UNKNOWN = 'unknown'

Runbook dataclass

Python
Runbook(name: str, description: str = '', steps: tuple[RunbookStep, ...] = ())

name instance-attribute

Python
name: str

description class-attribute instance-attribute

Python
description: str = ''

steps class-attribute instance-attribute

Python
steps: tuple[RunbookStep, ...] = ()

RunbookStep dataclass

Python
RunbookStep(name: str, action: str, timeout_s: float = 30.0, on_failure: str = 'stop')

name instance-attribute

Python
name: str

action instance-attribute

Python
action: str

timeout_s class-attribute instance-attribute

Python
timeout_s: float = 30.0

on_failure class-attribute instance-attribute

Python
on_failure: str = 'stop'

ServiceHealth dataclass

Python
ServiceHealth(name: str, status: HealthStatus = UNKNOWN, latency_ms: float = 0.0, message: str = '', metadata: dict[str, str] = dict())

name instance-attribute

Python
name: str

status class-attribute instance-attribute

Python
status: HealthStatus = UNKNOWN

latency_ms class-attribute instance-attribute

Python
latency_ms: float = 0.0

message class-attribute instance-attribute

Python
message: str = ''

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

is_healthy property

Python
is_healthy: bool

Domain · Enums

DeployerKind

Bases: str, Enum

DRY_RUN class-attribute instance-attribute

Python
DRY_RUN = 'dry_run'

PROCESS class-attribute instance-attribute

Python
PROCESS = 'process'

DOCKER class-attribute instance-attribute

Python
DOCKER = 'docker'

K8S class-attribute instance-attribute

Python
K8S = 'k8s'

Domain · Exceptions

AlertDeliveryError

Bases: OpsError

DeploymentError

Python
DeploymentError(name: str, message: str)

Bases: OpsError

Source code in apogee_ai_ops/domain/exceptions/ops_exceptions.py
Python
def __init__(self, name: str, message: str) -> None:
    super().__init__(f"Deployment {name!r} failed: {message}")
    self.name = name

name instance-attribute

Python
name = name

HealthcheckError

Bases: OpsError

OpsError

Bases: Exception

Base for apogee-ai-ops errors.

Domain · Protocols (ports)

IAlertSink

Bases: Protocol

name instance-attribute

Python
name: str

emit async

Python
emit(alert: Alert) -> None
Source code in apogee_ai_ops/domain/services/i_alert_sink.py
Python
async def emit(self, alert: Alert) -> None: ...

IDeployer

Bases: Protocol

name instance-attribute

Python
name: str

deploy async

Python
deploy(deployment: Deployment) -> Deployment
Source code in apogee_ai_ops/domain/services/i_deployer.py
Python
async def deploy(self, deployment: Deployment) -> Deployment: ...

rollback async

Python
rollback(name: str) -> None
Source code in apogee_ai_ops/domain/services/i_deployer.py
Python
async def rollback(self, name: str) -> None: ...

status async

Python
status(name: str) -> Deployment
Source code in apogee_ai_ops/domain/services/i_deployer.py
Python
async def status(self, name: str) -> Deployment: ...

IHealthcheck

Bases: Protocol

name instance-attribute

Python
name: str

check async

Python
check(target: str) -> ServiceHealth
Source code in apogee_ai_ops/domain/services/i_healthcheck.py
Python
async def check(self, target: str) -> ServiceHealth: ...

IRunbookRunner

Bases: Protocol

run async

Python
run(runbook: Runbook) -> dict
Source code in apogee_ai_ops/domain/services/i_runbook_runner.py
Python
async def run(self, runbook: Runbook) -> dict: ...

Infrastructure

CompositeHealthcheck

Python
CompositeHealthcheck(members: list[tuple[object, str]])

Aggregates multiple healthchecks against multiple targets in parallel.

Source code in apogee_ai_ops/infrastructure/health/composite_healthcheck.py
Python
def __init__(self, members: list[tuple[object, str]]) -> None:
    if not members:
        raise ValueError("at least one (checker, target) pair is required")
    self._members = members

name class-attribute instance-attribute

Python
name = 'composite'

check async

Python
check(target: str = 'composite') -> ServiceHealth
Source code in apogee_ai_ops/infrastructure/health/composite_healthcheck.py
Python
async def check(self, target: str = "composite") -> ServiceHealth:
    results = await asyncio.gather(
        *[checker.check(t) for checker, t in self._members]
    )
    if all(r.is_healthy for r in results):
        status = HealthStatus.HEALTHY
    elif any(r.status is HealthStatus.UNHEALTHY for r in results):
        status = HealthStatus.UNHEALTHY
    else:
        status = HealthStatus.DEGRADED
    avg_latency = sum(r.latency_ms for r in results) / len(results)
    return ServiceHealth(
        name=target,
        status=status,
        latency_ms=avg_latency,
        metadata={"members": str(len(results))},
    )

ConsoleAlertSink

Python
ConsoleAlertSink(stream: TextIO | None = None)
Source code in apogee_ai_ops/infrastructure/alerts/console_alert_sink.py
Python
def __init__(self, stream: TextIO | None = None) -> None:
    self._stream = stream or sys.stderr
    self.events: list[Alert] = []

name class-attribute instance-attribute

Python
name = 'console'

events instance-attribute

Python
events: list[Alert] = []

emit async

Python
emit(alert: Alert) -> None
Source code in apogee_ai_ops/infrastructure/alerts/console_alert_sink.py
Python
async def emit(self, alert: Alert) -> None:
    self.events.append(alert)
    line = (
        f"[{alert.severity.value.upper():8s}] "
        f"{alert.source}: {alert.title} — {alert.message}"
    )
    print(line, file=self._stream)

DeployerRegistry

Python
DeployerRegistry()
Source code in apogee_ai_ops/infrastructure/registries/deployer_registry.py
Python
def __init__(self) -> None:
    self._deployers: dict[str, object] = {
        "dry_run": DryRunDeployer(),
        "process": ProcessDeployer(),
    }

register

Python
register(name: str, deployer) -> None
Source code in apogee_ai_ops/infrastructure/registries/deployer_registry.py
Python
def register(self, name: str, deployer) -> None:
    self._deployers[name] = deployer

get

Python
get(name: str)
Source code in apogee_ai_ops/infrastructure/registries/deployer_registry.py
Python
def get(self, name: str):
    if name not in self._deployers:
        raise KeyError(f"unknown deployer: {name}")
    return self._deployers[name]

list

Python
list() -> list[str]
Source code in apogee_ai_ops/infrastructure/registries/deployer_registry.py
Python
def list(self) -> list[str]:
    return list(self._deployers.keys())

DockerDeployer

Python
DockerDeployer()

Lazy Docker SDK adapter — install via extras=docker.

Source code in apogee_ai_ops/infrastructure/deployers/docker_deployer.py
Python
def __init__(self) -> None:
    self._client = None
    self._containers: dict[str, str] = {}  # name -> container_id

name class-attribute instance-attribute

Python
name = 'docker'

deploy async

Python
deploy(deployment: Deployment) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/docker_deployer.py
Python
async def deploy(self, deployment: Deployment) -> Deployment:
    self._ensure_client()
    if not deployment.image:
        raise DeploymentError(deployment.name, "image required")
    try:
        container = self._client.containers.run(  # type: ignore[union-attr]
            deployment.image,
            command=list(deployment.command) or None,
            environment=dict(deployment.env),
            detach=True,
            name=deployment.name,
        )
    except Exception as exc:  # pragma: no cover
        raise DeploymentError(deployment.name, str(exc)) from exc
    self._containers[deployment.name] = container.id
    return replace(deployment, status=DeploymentStatus.RUNNING)

rollback async

Python
rollback(name: str) -> None
Source code in apogee_ai_ops/infrastructure/deployers/docker_deployer.py
Python
async def rollback(self, name: str) -> None:
    self._ensure_client()
    cid = self._containers.get(name)
    if cid is None:
        raise DeploymentError(name, "not deployed")
    try:
        container = self._client.containers.get(cid)  # type: ignore[union-attr]
        container.stop(timeout=5)
        container.remove()
    except Exception as exc:  # pragma: no cover
        raise DeploymentError(name, str(exc)) from exc

status async

Python
status(name: str) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/docker_deployer.py
Python
async def status(self, name: str) -> Deployment:
    self._ensure_client()
    if name not in self._containers:
        raise DeploymentError(name, "not deployed")
    return Deployment(name=name, image="?", status=DeploymentStatus.RUNNING)

DryRunDeployer

Python
DryRunDeployer()

Deployer that records calls without performing real actions.

CI-safe — never touches docker, k8s or processes.

Source code in apogee_ai_ops/infrastructure/deployers/dry_run_deployer.py
Python
def __init__(self) -> None:
    self._state: dict[str, Deployment] = {}
    self.calls: list[tuple[str, str]] = []

name class-attribute instance-attribute

Python
name = 'dry_run'

calls instance-attribute

Python
calls: list[tuple[str, str]] = []

deploy async

Python
deploy(deployment: Deployment) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/dry_run_deployer.py
Python
async def deploy(self, deployment: Deployment) -> Deployment:
    deployed = replace(deployment, status=DeploymentStatus.SUCCEEDED)
    self._state[deployment.name] = deployed
    self.calls.append(("deploy", deployment.name))
    return deployed

rollback async

Python
rollback(name: str) -> None
Source code in apogee_ai_ops/infrastructure/deployers/dry_run_deployer.py
Python
async def rollback(self, name: str) -> None:
    if name not in self._state:
        raise DeploymentError(name, "not deployed")
    self._state[name] = replace(
        self._state[name], status=DeploymentStatus.ROLLED_BACK
    )
    self.calls.append(("rollback", name))

status async

Python
status(name: str) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/dry_run_deployer.py
Python
async def status(self, name: str) -> Deployment:
    if name not in self._state:
        raise DeploymentError(name, "not deployed")
    return self._state[name]

HttpHealthcheck

Python
HttpHealthcheck(timeout_s: float = 5.0, expected_status: int = 200)
Source code in apogee_ai_ops/infrastructure/health/http_healthcheck.py
Python
def __init__(self, timeout_s: float = 5.0, expected_status: int = 200) -> None:
    self._timeout = timeout_s
    self._expected = expected_status

name class-attribute instance-attribute

Python
name = 'http'

check async

Python
check(target: str) -> ServiceHealth
Source code in apogee_ai_ops/infrastructure/health/http_healthcheck.py
Python
async def check(self, target: str) -> ServiceHealth:
    start = time.perf_counter()
    status, message = await asyncio.to_thread(self._check_sync, target)
    latency = (time.perf_counter() - start) * 1000.0
    if status == self._expected:
        return ServiceHealth(
            name=target, status=HealthStatus.HEALTHY,
            latency_ms=latency, metadata={"http_status": str(status)},
        )
    if 200 <= status < 500:
        return ServiceHealth(
            name=target, status=HealthStatus.DEGRADED,
            latency_ms=latency, message=message,
            metadata={"http_status": str(status)},
        )
    return ServiceHealth(
        name=target, status=HealthStatus.UNHEALTHY,
        latency_ms=latency, message=message,
        metadata={"http_status": str(status)},
    )

JsonFileAlertSink

Python
JsonFileAlertSink(path: str | Path)

Appends one JSON object per line to a file. Atomic per-line writes.

Source code in apogee_ai_ops/infrastructure/alerts/json_file_alert_sink.py
Python
def __init__(self, path: str | Path) -> None:
    self._path = Path(path)
    self._path.parent.mkdir(parents=True, exist_ok=True)

name class-attribute instance-attribute

Python
name = 'json_file'

emit async

Python
emit(alert: Alert) -> None
Source code in apogee_ai_ops/infrastructure/alerts/json_file_alert_sink.py
Python
async def emit(self, alert: Alert) -> None:
    payload = {
        "title": alert.title,
        "message": alert.message,
        "severity": alert.severity.value,
        "source": alert.source,
        "timestamp_s": alert.timestamp_s,
        "metadata": dict(alert.metadata),
    }
    with self._path.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(payload) + "\n")

K8sDeployer

Python
K8sDeployer(namespace: str = 'default')

Lazy Kubernetes adapter — install via extras=k8s.

Source code in apogee_ai_ops/infrastructure/deployers/k8s_deployer.py
Python
def __init__(self, namespace: str = "default") -> None:
    self._namespace = namespace
    self._client = None
    self._apps = None

name class-attribute instance-attribute

Python
name = 'k8s'

deploy async

Python
deploy(deployment: Deployment) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/k8s_deployer.py
Python
async def deploy(self, deployment: Deployment) -> Deployment:
    self._ensure_client()
    from kubernetes import client  # type: ignore

    body = client.V1Deployment(
        metadata=client.V1ObjectMeta(name=deployment.name),
        spec=client.V1DeploymentSpec(
            replicas=deployment.replicas,
            selector=client.V1LabelSelector(
                match_labels={"app": deployment.name}
            ),
            template=client.V1PodTemplateSpec(
                metadata=client.V1ObjectMeta(labels={"app": deployment.name}),
                spec=client.V1PodSpec(containers=[
                    client.V1Container(
                        name=deployment.name,
                        image=deployment.image,
                        command=list(deployment.command) or None,
                        env=[client.V1EnvVar(name=k, value=v)
                             for k, v in deployment.env.items()],
                    )
                ]),
            ),
        ),
    )
    try:
        self._apps.create_namespaced_deployment(  # type: ignore[union-attr]
            namespace=self._namespace, body=body
        )
    except Exception as exc:  # pragma: no cover
        raise DeploymentError(deployment.name, str(exc)) from exc
    return replace(deployment, status=DeploymentStatus.RUNNING)

rollback async

Python
rollback(name: str) -> None
Source code in apogee_ai_ops/infrastructure/deployers/k8s_deployer.py
Python
async def rollback(self, name: str) -> None:
    self._ensure_client()
    try:
        self._apps.delete_namespaced_deployment(  # type: ignore[union-attr]
            name=name, namespace=self._namespace
        )
    except Exception as exc:  # pragma: no cover
        raise DeploymentError(name, str(exc)) from exc

status async

Python
status(name: str) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/k8s_deployer.py
Python
async def status(self, name: str) -> Deployment:
    self._ensure_client()
    try:
        d = self._apps.read_namespaced_deployment(  # type: ignore[union-attr]
            name=name, namespace=self._namespace
        )
    except Exception as exc:  # pragma: no cover
        raise DeploymentError(name, str(exc)) from exc
    ready = (d.status.ready_replicas or 0) >= (d.spec.replicas or 1)
    return Deployment(
        name=name,
        image=d.spec.template.spec.containers[0].image or "",
        replicas=d.spec.replicas or 1,
        status=DeploymentStatus.SUCCEEDED if ready else DeploymentStatus.PENDING,
    )

ProcessDeployer

Python
ProcessDeployer()

Spawns deployments as local subprocesses. Useful for dev/stub agents.

Source code in apogee_ai_ops/infrastructure/deployers/process_deployer.py
Python
def __init__(self) -> None:
    self._processes: dict[str, asyncio.subprocess.Process] = {}
    self._state: dict[str, Deployment] = {}

name class-attribute instance-attribute

Python
name = 'process'

deploy async

Python
deploy(deployment: Deployment) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/process_deployer.py
Python
async def deploy(self, deployment: Deployment) -> Deployment:
    if not deployment.command:
        raise DeploymentError(
            deployment.name, "command required for process deployer"
        )
    env = {**os.environ, **deployment.env}
    try:
        proc = await asyncio.create_subprocess_exec(
            *deployment.command,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
    except (OSError, FileNotFoundError) as exc:
        raise DeploymentError(deployment.name, str(exc)) from exc
    self._processes[deployment.name] = proc
    deployed = replace(deployment, status=DeploymentStatus.RUNNING)
    self._state[deployment.name] = deployed
    return deployed

rollback async

Python
rollback(name: str) -> None
Source code in apogee_ai_ops/infrastructure/deployers/process_deployer.py
Python
async def rollback(self, name: str) -> None:
    proc = self._processes.get(name)
    if proc is None:
        raise DeploymentError(name, "not deployed")
    if proc.returncode is None:
        try:
            proc.send_signal(signal.SIGTERM)
        except ProcessLookupError:
            pass
        try:
            await asyncio.wait_for(proc.wait(), timeout=5.0)
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
    self._state[name] = replace(
        self._state[name], status=DeploymentStatus.ROLLED_BACK
    )

status async

Python
status(name: str) -> Deployment
Source code in apogee_ai_ops/infrastructure/deployers/process_deployer.py
Python
async def status(self, name: str) -> Deployment:
    if name not in self._state:
        raise DeploymentError(name, "not deployed")
    proc = self._processes.get(name)
    if proc is not None and proc.returncode is not None:
        current = (
            DeploymentStatus.SUCCEEDED if proc.returncode == 0 else DeploymentStatus.FAILED
        )
        self._state[name] = replace(self._state[name], status=current)
    return self._state[name]

SlackAlertSink

Python
SlackAlertSink(channel: str, token: str)

Lazy slack-sdk adapter — install via extras=slack.

Source code in apogee_ai_ops/infrastructure/alerts/slack_alert_sink.py
Python
def __init__(self, channel: str, token: str) -> None:
    self._channel = channel
    self._token = token
    self._client = None

name class-attribute instance-attribute

Python
name = 'slack'

emit async

Python
emit(alert: Alert) -> None
Source code in apogee_ai_ops/infrastructure/alerts/slack_alert_sink.py
Python
async def emit(self, alert: Alert) -> None:
    self._ensure_client()
    emoji = _SEVERITY_EMOJI.get(alert.severity.value, ":bell:")
    text = f"{emoji} *{alert.title}*\n{alert.message}\n_source: {alert.source}_"
    try:
        await self._client.chat_postMessage(  # type: ignore[union-attr]
            channel=self._channel, text=text,
        )
    except Exception as exc:  # pragma: no cover
        raise AlertDeliveryError(str(exc)) from exc

TcpHealthcheck

Python
TcpHealthcheck(timeout_s: float = 3.0)

Pings host:port via asyncio sockets.

Source code in apogee_ai_ops/infrastructure/health/tcp_healthcheck.py
Python
def __init__(self, timeout_s: float = 3.0) -> None:
    self._timeout = timeout_s

name class-attribute instance-attribute

Python
name = 'tcp'

check async

Python
check(target: str) -> ServiceHealth
Source code in apogee_ai_ops/infrastructure/health/tcp_healthcheck.py
Python
async def check(self, target: str) -> ServiceHealth:
    if ":" not in target:
        return ServiceHealth(
            name=target, status=HealthStatus.UNKNOWN,
            message="target must be host:port",
        )
    host, _, port_str = target.rpartition(":")
    try:
        port = int(port_str)
    except ValueError:
        return ServiceHealth(
            name=target, status=HealthStatus.UNKNOWN,
            message=f"invalid port: {port_str!r}",
        )
    start = time.perf_counter()
    try:
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port), timeout=self._timeout,
        )
    except (OSError, asyncio.TimeoutError) as exc:
        return ServiceHealth(
            name=target, status=HealthStatus.UNHEALTHY,
            latency_ms=(time.perf_counter() - start) * 1000.0,
            message=str(exc),
        )
    writer.close()
    try:
        await writer.wait_closed()
    except Exception:  # noqa: BLE001
        pass
    return ServiceHealth(
        name=target, status=HealthStatus.HEALTHY,
        latency_ms=(time.perf_counter() - start) * 1000.0,
    )