Lars
fe1edb84f8
All checks were successful
Deploy Development / deploy (push) Successful in 50s
Test Suite / pytest-backend (push) Successful in 3m49s
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 19s
Test Suite / playwright-smoke (push) Successful in 12s
apply_steering_event an Action done, Gate verify/reopen und Blocker resolve; lifecycle-Ketten methodenaware via slot_map; continuous_product slot_map; pytest v0.2. Co-authored-by: Cursor <cursoragent@cursor.com>
82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""Minimal TenantContext for service-layer steering calls."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from db import get_connection
|
|
from psycopg2.extras import RealDictCursor
|
|
from rights_registry import load_grants_for_roles
|
|
from tenant_context import TenantContext
|
|
|
|
|
|
def resolve_minimal_tenant_context(
|
|
*,
|
|
tenant_id: str,
|
|
user_id: Optional[str] = None,
|
|
) -> TenantContext:
|
|
"""Build TenantContext for kernel calls from OM services (no HTTP session)."""
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute(
|
|
"SELECT id, slug, name FROM tenants WHERE id = %s",
|
|
(tenant_id,),
|
|
)
|
|
tenant = cur.fetchone()
|
|
if not tenant:
|
|
raise ValueError("Tenant nicht gefunden")
|
|
|
|
user_row = None
|
|
membership_role: str | None = None
|
|
actor_id: str | None = None
|
|
if user_id:
|
|
cur.execute(
|
|
"""
|
|
SELECT u.id, u.email, u.display_name, u.portal_role, tm.tenant_role
|
|
FROM users u
|
|
LEFT JOIN tenant_memberships tm
|
|
ON tm.user_id = u.id AND tm.tenant_id = %s AND tm.is_active = TRUE
|
|
WHERE u.id = %s
|
|
""",
|
|
(tenant_id, user_id),
|
|
)
|
|
user_row = cur.fetchone()
|
|
if user_row:
|
|
membership_role = user_row.get("tenant_role")
|
|
cur.execute(
|
|
"""
|
|
SELECT id FROM actors
|
|
WHERE tenant_id = %s AND user_id = %s AND actor_type = 'human'
|
|
LIMIT 1
|
|
""",
|
|
(tenant_id, user_id),
|
|
)
|
|
actor = cur.fetchone()
|
|
actor_id = str(actor["id"]) if actor else None
|
|
finally:
|
|
conn.close()
|
|
|
|
portal_role = user_row["portal_role"] if user_row else "user"
|
|
caps = frozenset(
|
|
load_grants_for_roles(
|
|
portal_role=portal_role,
|
|
tenant_role=membership_role,
|
|
)
|
|
)
|
|
|
|
return TenantContext(
|
|
user_id=str(user_row["id"]) if user_row else "",
|
|
email=user_row["email"] if user_row else "",
|
|
display_name=user_row["display_name"] if user_row else "",
|
|
portal_role=portal_role,
|
|
tenant_id=str(tenant["id"]),
|
|
tenant_slug=tenant["slug"],
|
|
tenant_name=tenant["name"],
|
|
tenant_role=membership_role,
|
|
actor_id=actor_id,
|
|
actor_type="human" if actor_id else None,
|
|
session_token="steering-service",
|
|
capabilities=caps,
|
|
)
|