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>
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
"""Filtering, search and sorting specifications — Phase 11.9."""
|
|
from dataclasses import dataclass
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import asc, desc, or_
|
|
|
|
from app.models.integrations import (
|
|
ExperienceConsumerConnectorDispatch,
|
|
ExperienceConsumerConnectorRegistration,
|
|
ExperienceNotifyDispatch,
|
|
ExperienceWidgetDefinition,
|
|
ExperienceWidgetInstance,
|
|
)
|
|
|
|
CONNECTOR_REGISTRATION_SORT = frozenset(
|
|
{"created_at", "updated_at", "code", "name", "status", "kind"}
|
|
)
|
|
CONNECTOR_DISPATCH_SORT = frozenset(
|
|
{"created_at", "updated_at", "status", "event_type", "direction"}
|
|
)
|
|
WIDGET_DEFINITION_SORT = frozenset(
|
|
{"created_at", "updated_at", "code", "name", "status", "widget_kind"}
|
|
)
|
|
WIDGET_INSTANCE_SORT = frozenset(
|
|
{"created_at", "updated_at", "code", "name", "status", "embed_ref"}
|
|
)
|
|
NOTIFY_DISPATCH_SORT = frozenset(
|
|
{"created_at", "updated_at", "status", "channel", "template_key"}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class IntegrationListSpec:
|
|
workspace_id: UUID | None = None
|
|
site_id: UUID | None = None
|
|
page_id: UUID | None = None
|
|
status: str | None = None
|
|
kind: str | None = None
|
|
widget_kind: str | None = None
|
|
registration_id: UUID | None = None
|
|
q: str | None = None
|
|
sort_by: str = "created_at"
|
|
sort_dir: str = "desc"
|
|
model = None
|
|
sort_fields = frozenset()
|
|
|
|
def filter_clauses(self):
|
|
m = self.model
|
|
clauses = []
|
|
for field in (
|
|
"workspace_id",
|
|
"site_id",
|
|
"page_id",
|
|
"status",
|
|
"kind",
|
|
"widget_kind",
|
|
"registration_id",
|
|
):
|
|
value = getattr(self, field)
|
|
if value is not None and hasattr(m, field):
|
|
clauses.append(getattr(m, field) == value)
|
|
if self.q:
|
|
term = f"%{self.q.strip()}%"
|
|
searchable = [
|
|
getattr(m, name).ilike(term)
|
|
for name in (
|
|
"code",
|
|
"name",
|
|
"description",
|
|
"event_type",
|
|
"embed_ref",
|
|
"template_key",
|
|
"channel",
|
|
)
|
|
if hasattr(m, name)
|
|
]
|
|
if searchable:
|
|
clauses.append(or_(*searchable))
|
|
return clauses
|
|
|
|
def apply(self, stmt):
|
|
clauses = self.filter_clauses()
|
|
if clauses:
|
|
stmt = stmt.where(*clauses)
|
|
key = self.sort_by if self.sort_by in self.sort_fields else "created_at"
|
|
column = getattr(self.model, key)
|
|
return stmt.order_by(asc(column) if self.sort_dir.lower() == "asc" else desc(column))
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ConsumerConnectorRegistrationListSpec(IntegrationListSpec):
|
|
model = ExperienceConsumerConnectorRegistration
|
|
sort_fields = CONNECTOR_REGISTRATION_SORT
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ConsumerConnectorDispatchListSpec(IntegrationListSpec):
|
|
model = ExperienceConsumerConnectorDispatch
|
|
sort_fields = CONNECTOR_DISPATCH_SORT
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class WidgetDefinitionListSpec(IntegrationListSpec):
|
|
model = ExperienceWidgetDefinition
|
|
sort_fields = WIDGET_DEFINITION_SORT
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class WidgetInstanceListSpec(IntegrationListSpec):
|
|
model = ExperienceWidgetInstance
|
|
sort_fields = WIDGET_INSTANCE_SORT
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class NotifyDispatchListSpec(IntegrationListSpec):
|
|
model = ExperienceNotifyDispatch
|
|
sort_fields = NOTIFY_DISPATCH_SORT
|