Lars
564008be13
Some checks failed
Deploy Development / deploy (push) Successful in 38s
Test Suite / pytest-backend (push) Failing after 7s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Co-authored-by: Cursor <cursoragent@cursor.com>
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
"""Actor creation helpers — User and Actor remain separate concepts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Literal, Optional
|
|
|
|
from psycopg2.extras import RealDictCursor
|
|
|
|
from db import get_connection
|
|
|
|
ActorType = Literal["human", "agent", "working_group", "external_system"]
|
|
|
|
|
|
def create_actor(
|
|
*,
|
|
tenant_id: str,
|
|
actor_type: ActorType,
|
|
name: str,
|
|
user_id: Optional[str] = None,
|
|
) -> dict[str, Any]:
|
|
if actor_type == "human" and not user_id:
|
|
raise ValueError("Human actors require user_id")
|
|
if actor_type != "human" and user_id:
|
|
raise ValueError("Non-human actors must not have user_id")
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO actors (tenant_id, actor_type, name, user_id)
|
|
VALUES (%s, %s, %s, %s)
|
|
RETURNING id, tenant_id, actor_type, name, user_id, is_active, created_at
|
|
""",
|
|
(tenant_id, actor_type, name, user_id),
|
|
)
|
|
row = dict(cur.fetchone())
|
|
conn.commit()
|
|
for key in ("id", "tenant_id", "user_id"):
|
|
if row.get(key):
|
|
row[key] = str(row[key])
|
|
return row
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_human_actor(tenant_id: str, user_id: str) -> Optional[dict[str, Any]]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, tenant_id, actor_type, name, user_id, is_active, created_at
|
|
FROM actors
|
|
WHERE tenant_id = %s AND user_id = %s AND actor_type = 'human'
|
|
""",
|
|
(tenant_id, user_id),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
result = dict(row)
|
|
for key in ("id", "tenant_id", "user_id"):
|
|
if result.get(key):
|
|
result[key] = str(result[key])
|
|
return result
|
|
finally:
|
|
conn.close()
|