61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
from __future__ import annotations
|
|
from typing import Any, Optional
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class Resource(BaseModel):
|
|
resourceName: str
|
|
resourceIdentifier: str
|
|
alias: Optional[str] = None
|
|
isActive: bool
|
|
|
|
|
|
class EventTypeChannelPreference(BaseModel):
|
|
eventTypeChannelId: str
|
|
channelId: int
|
|
name: str
|
|
isMandatory: bool = False
|
|
resources: list[Resource] = Field(default_factory=list)
|
|
|
|
|
|
class EventTypePreference(BaseModel):
|
|
eventTypeId: int
|
|
name: str
|
|
subName: Optional[str] = None
|
|
prettyName: str
|
|
eventTypeChannels: list[EventTypeChannelPreference] = Field(default_factory=list)
|
|
|
|
|
|
class OvPayToken(BaseModel):
|
|
ovPayTokenId: int
|
|
alias: Optional[str] = None
|
|
|
|
|
|
class NotificationSubscriptionPreference(BaseModel):
|
|
notificationSubscriptionId: str
|
|
customerProfileId: int
|
|
ovPayToken: Optional[OvPayToken] = None
|
|
isActive: bool
|
|
eventTypes: list[EventTypePreference] = Field(default_factory=list)
|
|
|
|
|
|
class NotificationCategoryPreference(BaseModel):
|
|
notificationCategoryId: int
|
|
name: str
|
|
groupName: Optional[str] = None
|
|
notificationSubscriptions: list[NotificationSubscriptionPreference] = Field(default_factory=list)
|
|
|
|
|
|
class CustomerNotificationPreferencesResponse(BaseModel):
|
|
notificationCategories: list[NotificationCategoryPreference] = Field(default_factory=list)
|
|
|
|
|
|
def latest_activity_is_active(subscription: dict[str, Any]) -> bool:
|
|
"""CRUD returns subscriptionActivities; the most recent/current one determines active state."""
|
|
activities = subscription.get("subscriptionActivities") or []
|
|
if not activities:
|
|
return False
|
|
# CRUD activityLimit defaults to 1; when more are supplied, sort defensively on timestamp.
|
|
latest = sorted(activities, key=lambda a: a.get("timestamp", ""), reverse=True)[0]
|
|
return bool(latest.get("isActive"))
|