Lars
95feba6204
All checks were successful
Deploy Development / deploy (push) Successful in 44s
Test Suite / pytest-backend (push) Successful in 1m19s
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 11s
Persistierter Standard Lifecycle pro Initiative, backend/steering/ Skeleton, Attention-Refactor und Hook-Dispatch als Basis für Method Registry. Co-authored-by: Cursor <cursoragent@cursor.com>
36 lines
1003 B
Python
36 lines
1003 B
Python
"""SteeringContext read helpers — AP1.0."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from services import steering_context as sc_service
|
|
from steering.lifecycle.states import lifecycle_label
|
|
from tenant_context import TenantContext
|
|
|
|
|
|
def get_steering_context_dto(
|
|
ctx: TenantContext, *, initiative_id: str
|
|
) -> Optional[dict[str, Any]]:
|
|
row = sc_service.get_steering_context(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
if not row:
|
|
row = sc_service.ensure_steering_context(
|
|
tenant_id=ctx.tenant_id,
|
|
initiative_id=initiative_id,
|
|
)
|
|
return {
|
|
**row,
|
|
"lifecycle_label": lifecycle_label(row["lifecycle_state"]),
|
|
}
|
|
|
|
|
|
def get_or_create_for_initiative(
|
|
ctx: TenantContext, *, initiative_id: str
|
|
) -> dict[str, Any]:
|
|
dto = get_steering_context_dto(ctx, initiative_id=initiative_id)
|
|
if not dto:
|
|
raise ValueError("SteeringContext nicht verfügbar")
|
|
return dto
|