38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
|
"""Initiative read-models for workspace."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Optional
|
||
|
|
|
||
|
|
from psycopg2.extras import RealDictCursor
|
||
|
|
|
||
|
|
from db import get_connection
|
||
|
|
from services.initiatives import _serialize_row
|
||
|
|
from tenant_context import TenantContext
|
||
|
|
|
||
|
|
|
||
|
|
def get_active_initiatives(
|
||
|
|
ctx: TenantContext,
|
||
|
|
*,
|
||
|
|
limit: Optional[int] = None,
|
||
|
|
) -> list[dict[str, Any]]:
|
||
|
|
"""Active and paused initiatives in the current tenant."""
|
||
|
|
conn = get_connection()
|
||
|
|
try:
|
||
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
|
|
sql = """
|
||
|
|
SELECT id, tenant_id, title, goal, status, priority,
|
||
|
|
owner_actor_id, created_at, updated_at
|
||
|
|
FROM initiatives
|
||
|
|
WHERE tenant_id = %s AND status IN ('active', 'paused')
|
||
|
|
ORDER BY updated_at DESC, title
|
||
|
|
"""
|
||
|
|
params: list[Any] = [ctx.tenant_id]
|
||
|
|
if limit is not None:
|
||
|
|
sql += " LIMIT %s"
|
||
|
|
params.append(limit)
|
||
|
|
cur.execute(sql, params)
|
||
|
|
return [_serialize_row(dict(row)) for row in cur.fetchall()]
|
||
|
|
finally:
|
||
|
|
conn.close()
|