Lars
db06af0489
All checks were successful
Deploy Development / deploy (push) Successful in 47s
Test Suite / pytest-backend (push) Successful in 42s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 12s
Co-authored-by: Cursor <cursoragent@cursor.com>
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()
|