37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
|
"""fin→dev: resource pressure signals to dev-hub (State Hub progress)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from hub_core.events import ALERT_RAISED
|
||
|
|
|
||
|
|
from fin_hub.coupling.common import default_dev_hub_api_base, post_progress_event
|
||
|
|
from fin_hub.services.alerts import BudgetAlert
|
||
|
|
|
||
|
|
|
||
|
|
def emit_resource_pressure(
|
||
|
|
alerts: list[BudgetAlert],
|
||
|
|
*,
|
||
|
|
api_base: str | None = None,
|
||
|
|
domain_slug: str = "infotech",
|
||
|
|
) -> list[dict[str, Any]]:
|
||
|
|
base = api_base or default_dev_hub_api_base()
|
||
|
|
emitted: list[dict[str, Any]] = []
|
||
|
|
for alert in alerts:
|
||
|
|
if alert.code not in {"budget_pressure", "runway_below_threshold"}:
|
||
|
|
continue
|
||
|
|
result = post_progress_event(
|
||
|
|
api_base=base,
|
||
|
|
event_type=ALERT_RAISED,
|
||
|
|
summary=f"[fin→dev:{domain_slug}] {alert.summary}",
|
||
|
|
detail={
|
||
|
|
"source_hub": "fin-hub",
|
||
|
|
"target_hub": "dev-hub",
|
||
|
|
"signal": "resource_pressure",
|
||
|
|
"alert": alert.as_dict(),
|
||
|
|
},
|
||
|
|
author="fin-hub",
|
||
|
|
)
|
||
|
|
emitted.append(result)
|
||
|
|
return emitted
|