203671a7bf
Add the experience service with sites through analytics/AI hooks, migrations through 0011, and phase docs/manifests marking the track complete. Co-authored-by: Cursor <cursoragent@cursor.com>
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""Consumer connector & widget boundary validation — Phase 11.9."""
|
|
from typing import Any
|
|
|
|
from app.validators.foundation import ValidationError
|
|
from shared.exceptions import AppError
|
|
|
|
FORBIDDEN_OWNERSHIP_KEYS = frozenset(
|
|
{
|
|
"sms_provider",
|
|
"payamak",
|
|
"provider_registry",
|
|
"communication_engine",
|
|
"provider_adapter",
|
|
"message_queue",
|
|
"dlq",
|
|
"menu_catalog",
|
|
"product_catalog",
|
|
"crm_contact_id",
|
|
"member_id",
|
|
"booking_engine",
|
|
"schedule_engine",
|
|
"analytics_engine",
|
|
"ai_content_engine",
|
|
"model_inference",
|
|
"plan_engine",
|
|
"subscription_engine",
|
|
"billing_ledger",
|
|
"core_plan_id",
|
|
}
|
|
)
|
|
|
|
|
|
def _reject_forbidden(value: dict, field: str) -> None:
|
|
banned = FORBIDDEN_OWNERSHIP_KEYS.intersection(value.keys())
|
|
if banned:
|
|
raise ValidationError(
|
|
"کلیدهای مالکیت سرویسهای دیگر در payload مجاز نیست",
|
|
{"field": field, "forbidden_keys": sorted(banned)},
|
|
)
|
|
|
|
|
|
def validate_json_object(value: Any, field: str) -> dict | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, dict):
|
|
raise ValidationError(f"{field} باید object باشد", {"field": field})
|
|
_reject_forbidden(value, field)
|
|
return value
|
|
|
|
|
|
def validate_metadata(value: Any) -> dict | None:
|
|
return validate_json_object(value, "metadata_json")
|
|
|
|
|
|
def validate_config(value: Any) -> dict | None:
|
|
return validate_json_object(value, "config_json")
|
|
|
|
|
|
def validate_default_config(value: Any) -> dict | None:
|
|
return validate_json_object(value, "default_config_json")
|
|
|
|
|
|
def validate_payload(value: Any) -> dict | None:
|
|
return validate_json_object(value, "payload_json")
|
|
|
|
|
|
def ensure_transition(current, target, transitions: dict, label: str) -> None:
|
|
if current == target:
|
|
return
|
|
if target not in transitions.get(current, frozenset()):
|
|
raise AppError(
|
|
f"گذار وضعیت {label} مجاز نیست",
|
|
error_code=f"invalid_{label}_status_transition",
|
|
status_code=409,
|
|
details={"current": current.value, "target": target.value},
|
|
)
|