Lars
53047b6c16
All checks were successful
Deploy Development / deploy (push) Successful in 38s
Test Suite / pytest-backend (push) Successful in 18s
Test Suite / lint-backend (push) Successful in 1s
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 24s
Migration 005, Registry-Sync, Placeholder-Validation, capability-geschuetzte API, Tests und Abschlussbericht v0.1. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""Placeholder definitions registry with DB sync."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from db import get_connection
|
|
from feature_registry import VALID_CONTEXT_KINDS
|
|
|
|
VALID_VALUE_TYPES = frozenset({"string", "number", "boolean", "object", "array", "json"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlaceholderRegistration:
|
|
placeholder_key: str
|
|
context_kind: str
|
|
value_type: str
|
|
description: str
|
|
required: bool = False
|
|
source: str = "input"
|
|
|
|
|
|
_REGISTRY: list[PlaceholderRegistration] = []
|
|
|
|
|
|
def register_placeholder(registration: PlaceholderRegistration) -> None:
|
|
if not registration.placeholder_key or not registration.placeholder_key.strip():
|
|
raise ValueError("Placeholder key is required")
|
|
if registration.context_kind not in VALID_CONTEXT_KINDS:
|
|
raise ValueError(f"Unknown context_kind: {registration.context_kind}")
|
|
if registration.value_type not in VALID_VALUE_TYPES:
|
|
raise ValueError(f"Unknown value_type: {registration.value_type}")
|
|
if not registration.description or not registration.description.strip():
|
|
raise ValueError("Placeholder description is required")
|
|
key = (registration.placeholder_key, registration.context_kind)
|
|
if any((r.placeholder_key, r.context_kind) == key for r in _REGISTRY):
|
|
raise ValueError(f"Placeholder already registered: {key}")
|
|
_REGISTRY.append(registration)
|
|
|
|
|
|
def get_registered_placeholders() -> tuple[PlaceholderRegistration, ...]:
|
|
return tuple(_REGISTRY)
|
|
|
|
|
|
def clear_placeholder_registry_for_tests() -> None:
|
|
_REGISTRY.clear()
|
|
|
|
|
|
def sync_placeholders_to_db() -> int:
|
|
if not _REGISTRY:
|
|
print("[placeholder_registry] Keine Placeholder registriert — Sync übersprungen")
|
|
return 0
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
for reg in _REGISTRY:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO placeholder_definitions (
|
|
placeholder_key, context_kind, value_type, description,
|
|
required, source, is_active
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, TRUE)
|
|
ON CONFLICT (placeholder_key, context_kind) DO UPDATE SET
|
|
value_type = EXCLUDED.value_type,
|
|
description = EXCLUDED.description,
|
|
required = EXCLUDED.required,
|
|
source = EXCLUDED.source,
|
|
is_active = TRUE,
|
|
updated_at = NOW()
|
|
""",
|
|
(
|
|
reg.placeholder_key,
|
|
reg.context_kind,
|
|
reg.value_type,
|
|
reg.description,
|
|
reg.required,
|
|
reg.source,
|
|
),
|
|
)
|
|
conn.commit()
|
|
print(f"[placeholder_registry] Sync OK — {len(_REGISTRY)} Placeholder(s)")
|
|
return 0
|
|
except Exception as exc:
|
|
conn.rollback()
|
|
print(f"[placeholder_registry] Sync FAIL: {exc}")
|
|
return 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def load_placeholders_for_context(context_kind: str) -> list[dict]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT placeholder_key, value_type, description, required, source
|
|
FROM placeholder_definitions
|
|
WHERE context_kind = %s AND is_active = TRUE
|
|
ORDER BY placeholder_key
|
|
""",
|
|
(context_kind,),
|
|
)
|
|
return [
|
|
{
|
|
"placeholder_key": row[0],
|
|
"value_type": row[1],
|
|
"description": row[2],
|
|
"required": row[3],
|
|
"source": row[4],
|
|
}
|
|
for row in cur.fetchall()
|
|
]
|
|
finally:
|
|
conn.close()
|