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>
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""Lifecycle policies for Phase 11.9 aggregates."""
|
|
from app.models.types import (
|
|
ConnectorDispatchStatus as D,
|
|
ConsumerConnectorStatus as C,
|
|
NotifyDispatchStatus as N,
|
|
WidgetDefinitionStatus as WD,
|
|
WidgetInstanceStatus as WI,
|
|
)
|
|
from app.validators.integrations import ensure_transition
|
|
|
|
_TRANSITIONS = {
|
|
C: {
|
|
C.DRAFT: frozenset({C.ACTIVE, C.RETIRED}),
|
|
C.ACTIVE: frozenset({C.SUSPENDED, C.RETIRED}),
|
|
C.SUSPENDED: frozenset({C.ACTIVE, C.RETIRED}),
|
|
C.RETIRED: frozenset(),
|
|
},
|
|
D: {
|
|
D.PENDING: frozenset({D.ACKNOWLEDGED, D.FAILED, D.CANCELLED}),
|
|
D.ACKNOWLEDGED: frozenset(),
|
|
D.FAILED: frozenset({D.PENDING, D.CANCELLED}),
|
|
D.CANCELLED: frozenset(),
|
|
},
|
|
WD: {
|
|
WD.DRAFT: frozenset({WD.AVAILABLE, WD.RETIRED}),
|
|
WD.AVAILABLE: frozenset({WD.RETIRED}),
|
|
WD.RETIRED: frozenset(),
|
|
},
|
|
WI: {
|
|
WI.DRAFT: frozenset({WI.ACTIVE, WI.ARCHIVED}),
|
|
WI.ACTIVE: frozenset({WI.SUSPENDED, WI.ARCHIVED}),
|
|
WI.SUSPENDED: frozenset({WI.ACTIVE, WI.ARCHIVED}),
|
|
WI.ARCHIVED: frozenset(),
|
|
},
|
|
N: {
|
|
N.PENDING: frozenset({N.SENT, N.FAILED, N.CANCELLED}),
|
|
N.SENT: frozenset(),
|
|
N.FAILED: frozenset({N.PENDING, N.CANCELLED}),
|
|
N.CANCELLED: frozenset(),
|
|
},
|
|
}
|
|
|
|
|
|
class ConsumerConnectorRegistrationPolicy:
|
|
def assert_status_change(self, *, current, target):
|
|
ensure_transition(current, target, _TRANSITIONS[C], "consumer_connector_registration")
|
|
|
|
|
|
class ConsumerConnectorDispatchPolicy:
|
|
def assert_status_change(self, *, current, target):
|
|
ensure_transition(current, target, _TRANSITIONS[D], "consumer_connector_dispatch")
|
|
|
|
|
|
class WidgetDefinitionPolicy:
|
|
def assert_status_change(self, *, current, target):
|
|
ensure_transition(current, target, _TRANSITIONS[WD], "widget_definition")
|
|
|
|
|
|
class WidgetInstancePolicy:
|
|
def assert_status_change(self, *, current, target):
|
|
ensure_transition(current, target, _TRANSITIONS[WI], "widget_instance")
|
|
|
|
|
|
class NotifyDispatchPolicy:
|
|
def assert_status_change(self, *, current, target):
|
|
ensure_transition(current, target, _TRANSITIONS[N], "notify_dispatch")
|