50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
"""Optional project-local alert projection."""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from psycopg import AsyncConnection
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkbenchOperationsAvailability:
|
|
alerts: bool
|
|
schedules: bool = False
|
|
|
|
|
|
class WorkbenchOperationsRepository:
|
|
"""Read provisioned workbench records without creating or mutating storage."""
|
|
|
|
@staticmethod
|
|
async def availability(
|
|
conn: AsyncConnection,
|
|
) -> WorkbenchOperationsAvailability:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
"""
|
|
SELECT to_regclass('workbench.alerts') IS NOT NULL AS alerts
|
|
"""
|
|
)
|
|
row = await cur.fetchone()
|
|
return WorkbenchOperationsAvailability(
|
|
alerts=bool(row and row["alerts"]),
|
|
)
|
|
|
|
@staticmethod
|
|
async def list_alerts(
|
|
conn: AsyncConnection, project_id: str, *, limit: int = 100
|
|
) -> list[dict[str, Any]]:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
"""
|
|
SELECT document
|
|
FROM workbench.alerts
|
|
WHERE project_id = %s
|
|
AND status = 'active'
|
|
ORDER BY occurred_at DESC, alert_id
|
|
LIMIT %s
|
|
""",
|
|
(project_id, limit),
|
|
)
|
|
return [row["document"] for row in await cur.fetchall()]
|