Lars
4ad908bfbf
Some checks failed
Test Suite / pytest-backend (push) Waiting to run
Test Suite / lint-backend (push) Waiting to run
Test Suite / compose-smoke (push) Waiting to run
Test Suite / k6 /api/health Baseline (push) Blocked by required conditions
Test Suite / playwright-smoke (push) Blocked by required conditions
Deploy Development / deploy (push) Has been cancelled
Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Operating Context read model — AP2.3a."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from psycopg2.extras import RealDictCursor
|
|
|
|
from db import get_connection
|
|
from entity_archetypes.registry import get_ui_profile_json, resolve_ui_profile
|
|
from services.initiatives import get_initiative
|
|
from services.steering_context import ensure_steering_context, get_steering_context
|
|
from steering.methods.registry import get_method
|
|
|
|
|
|
def _load_ui_profile_from_db(archetype_key: str) -> Optional[dict[str, Any]]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT ui_profile_json
|
|
FROM entity_archetypes
|
|
WHERE archetype_key = %s
|
|
""",
|
|
(archetype_key,),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
profile = row.get("ui_profile_json")
|
|
if isinstance(profile, dict) and profile:
|
|
return dict(profile)
|
|
return None
|
|
except Exception:
|
|
return None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _resolve_ui_profile(archetype_key: str) -> dict[str, Any]:
|
|
db_profile = _load_ui_profile_from_db(archetype_key)
|
|
if db_profile:
|
|
return db_profile
|
|
seed_profile = get_ui_profile_json(archetype_key)
|
|
if seed_profile:
|
|
return seed_profile
|
|
return resolve_ui_profile(archetype_key)
|
|
|
|
|
|
def _resolve_data_slices(
|
|
*,
|
|
ui_profile: dict[str, Any],
|
|
method_key: str,
|
|
) -> list[str]:
|
|
"""Schnittmenge Archetyp-Default ∩ Method-Slices (Phase E erweitert)."""
|
|
archetype_slices = ui_profile.get("dataSlices") or []
|
|
method = get_method(method_key)
|
|
if method and getattr(method, "data_slices", None):
|
|
method_slices = set(method.data_slices)
|
|
return [s for s in archetype_slices if s in method_slices]
|
|
return list(archetype_slices)
|
|
|
|
|
|
def _method_capabilities(method_key: str) -> dict[str, Any]:
|
|
method = get_method(method_key)
|
|
if not method:
|
|
return {
|
|
"lifecycle_steps": [],
|
|
"next_action_strategy_key": "default",
|
|
}
|
|
return {
|
|
"lifecycle_steps": list(method.default_lifecycle_steps),
|
|
"next_action_strategy_key": method.next_action_strategy_key,
|
|
}
|
|
|
|
|
|
def get_operating_context(
|
|
*, tenant_id: str, initiative_id: str
|
|
) -> Optional[dict[str, Any]]:
|
|
initiative = get_initiative(tenant_id=tenant_id, initiative_id=initiative_id)
|
|
if not initiative:
|
|
return None
|
|
|
|
steering = get_steering_context(tenant_id=tenant_id, initiative_id=initiative_id)
|
|
if not steering:
|
|
steering = ensure_steering_context(
|
|
tenant_id=tenant_id, initiative_id=initiative_id
|
|
)
|
|
|
|
archetype_key = initiative["archetype_key"]
|
|
ui_profile = _resolve_ui_profile(archetype_key)
|
|
method_key = steering["method_key"]
|
|
metadata = steering.get("lifecycle_metadata") or {}
|
|
if isinstance(metadata, str):
|
|
metadata = {}
|
|
method_profile_key = metadata.get("method_profile_key")
|
|
data_slices = _resolve_data_slices(ui_profile=ui_profile, method_key=method_key)
|
|
|
|
return {
|
|
"initiative_id": initiative_id,
|
|
"archetype_key": archetype_key,
|
|
"method_key": method_key,
|
|
"method_profile_key": method_profile_key,
|
|
"ui_profile": ui_profile,
|
|
"data_slices": data_slices,
|
|
"method_capabilities": _method_capabilities(method_key),
|
|
}
|