First vibe coded example

src/python/notificatiesbeheren/notification-preferences-example/README.md

@@ -0,0 +1,72 @@
+# Notification preferences transformation service - Python example
+
+Reference implementation for the Service Engine `GET /customers/notificationpreferences` endpoint. It composes raw CRUD API data into the frontend-friendly response model from `SE-notifications.yaml`.
+
+## Supplied contracts used
+
+- `notifications-crud.yaml`
+  - `GET /notificationsubscriptions?customerProfileId={id}&expand=eventTypeChannel`
+  - `GET /notificationcategories?notificationCategoryId={id}&expand=eventTypeChannel`
+- `customers-crud-v2.yaml`
+  - `GET /ovpaytokens?ovPayTokenId={id}`
+  - `GET /devices?customerProfileId={id}`
+  - `GET /customers?customerProfileId={id}` for person/email resource data
+- `SE-notifications.yaml`
+  - `GET /customers/notificationpreferences`
+
+## Runtime options
+
+### Container App / local FastAPI
+
+```bash
+pip install -r requirements.txt
+export NOTIFICATIONS_CRUD_BASE_URL=https://your-notifications-crud-api.example.nl
+export CUSTOMERS_CRUD_BASE_URL=https://your-customers-crud-api.example.nl
+uvicorn app.entrypoints.fastapi_app:app --reload
+```
+
+Call:
+
+```bash
+curl -H "X-HTM-CUSTOMER-PROFILE-ID-HEADER: 42" \
+  "http://localhost:8000/customers/notificationpreferences?deviceId=5bedce29-af0c-4f3c-b182-2caa8a1f9377"
+```
+
+### Azure Function
+
+Use `app/entrypoints/azure_function.py` as the HTTP-trigger body. Keep the `app/services` and `app/domain` modules unchanged.
+
+## Design choice
+
+The implementation deliberately separates:
+
+- HTTP entrypoints: request/response only
+- CRUD client: raw REST calls only
+- Service layer: composition and transformation logic
+- Domain models: frontend/SE response shape
+
+This makes the same transformation reusable in Azure Functions, Container Apps, and tests.
+
+## Business rule interpretation
+
+For each customer subscription:
+
+1. Load customer `NotificationSubscriptions` with `NotificationPreferences` expanded to `EventTypeChannel`.
+2. Load the default category definition with all `EventTypes` and `EventTypeChannels`.
+3. Load customer resources for each channel resource type:
+   - `resourceNameId = 8` -> `GET /devices?customerProfileId={id}`
+   - `resourceNameId = 4` -> `GET /customers?customerProfileId={id}` and use the embedded `person` email data
+4. For every default `EventTypeChannel`, write out resources explicitly.
+5. If the channel is default-on and no preference exists, resource is active.
+6. If the channel is default-off and no preference exists, resource is inactive.
+7. A `NotificationPreference` row is interpreted as a deviation:
+   - default-on + preference -> inactive unless explicit `isActive` says otherwise
+   - default-off + preference -> active unless explicit `isActive` says otherwise
+8. A null `resourceIdentifier` applies to all resources for that EventTypeChannel.
+
+## Production hardening points
+
+- Add authentication propagation from the SE API to both CRUD APIs.
+- Decide whether category defaults should be cached across requests.
+- Decide whether missing category defaults should become `502 Bad Gateway`, partial data, or an empty category.
+- Confirm whether the real `NotificationPreference` has explicit `isActive`; this example supports it, but also works when preference existence alone means deviation.

src/python/notificatiesbeheren/notification-preferences-example/app/clients/crud_client.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+from typing import Any, Optional
+import httpx
+
+
+class CrudApiClient:
+    """Thin async client around the ABT CRUD APIs.
+
+    Keep business/transformation logic out of this class. The client only knows how
+    to call the source endpoints and unwrap the collection names from the CRUD
+    contracts.
+    """
+
+    def __init__(
+        self,
+        notifications_base_url: str,
+        customers_base_url: str,
+        timeout_seconds: float = 10.0,
+    ):
+        self.notifications_base_url = notifications_base_url.rstrip("/")
+        self.customers_base_url = customers_base_url.rstrip("/")
+        self.timeout_seconds = timeout_seconds
+
+    async def _get(
+        self,
+        base_url: str,
+        path: str,
+        params: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]:
+        async with httpx.AsyncClient(base_url=base_url, timeout=self.timeout_seconds) as client:
+            response = await client.get(path, params=params)
+            response.raise_for_status()
+            return response.json()
+
+    # notifications-crud.yaml
+    async def get_notification_subscriptions(self, customer_profile_id: int) -> list[dict[str, Any]]:
+        payload = await self._get(
+            self.notifications_base_url,
+            "/notificationsubscriptions",
+            params={"customerProfileId": customer_profile_id, "expand": "eventTypeChannel"},
+        )
+        return payload.get("notificationSubscriptions", [])
+
+    async def get_notification_category_defaults(self, notification_category_id: int) -> Optional[dict[str, Any]]:
+        payload = await self._get(
+            self.notifications_base_url,
+            "/notificationcategories",
+            params={"notificationCategoryId": notification_category_id, "expand": "eventTypeChannel"},
+        )
+        categories = payload.get("notificationCategories", [])
+        return categories[0] if categories else None
+
+    # customers-crud-v2.yaml
+    async def get_ovpay_token(self, ovpay_token_id: Optional[int]) -> Optional[dict[str, Any]]:
+        if ovpay_token_id is None:
+            return None
+        payload = await self._get(
+            self.customers_base_url,
+            "/ovpaytokens",
+            params={"ovPayTokenId": ovpay_token_id},
+        )
+        tokens = payload.get("ovPayTokens", [])
+        return tokens[0] if tokens else None
+
+    async def get_devices(self, customer_profile_id: int) -> list[dict[str, Any]]:
+        payload = await self._get(
+            self.customers_base_url,
+            "/devices",
+            params={"customerProfileId": customer_profile_id},
+        )
+        return payload.get("devices", [])
+
+    async def get_persons(self, customer_profile_id: int) -> list[dict[str, Any]]:
+        """Return customer/person resources used for email-like channels.
+
+        customers-crud-v2.yaml exposes person details through GET /customers,
+        filtered by customerProfileId. For the transformation service we normalize
+        the matching customer profile into one resource with the customer's email.
+        """
+        payload = await self._get(
+            self.customers_base_url,
+            "/customers",
+            params={"customerProfileId": customer_profile_id},
+        )
+        return payload.get("customers", [])
+
+
+# Backwards-compatible alias used by the service type hints.
+NotificationsCrudClient = CrudApiClient

src/python/notificatiesbeheren/notification-preferences-example/app/domain/models.py

@@ -0,0 +1,60 @@
+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"))

src/python/notificatiesbeheren/notification-preferences-example/app/entrypoints/azure_function.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+import azure.functions as func
+
+from app.clients.crud_client import NotificationsCrudClient
+from app.services.notification_preferences_service import NotificationPreferencesService
+from app.settings import settings
+
+
+async def main(req: func.HttpRequest) -> func.HttpResponse:
+    customer_profile_id = req.headers.get("X-HTM-CUSTOMER-PROFILE-ID-HEADER")
+    if not customer_profile_id:
+        return func.HttpResponse("Missing X-HTM-CUSTOMER-PROFILE-ID-HEADER", status_code=400)
+
+    service = NotificationPreferencesService(NotificationsCrudClient(settings.notifications_crud_base_url, settings.customers_crud_base_url))
+    result = await service.get_customer_notification_preferences(
+        customer_profile_id=int(customer_profile_id),
+        device_id=req.params.get("deviceId"),
+    )
+    return func.HttpResponse(result.model_dump_json(exclude_none=True), mimetype="application/json", status_code=200)

src/python/notificatiesbeheren/notification-preferences-example/app/entrypoints/fastapi_app.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+from fastapi import FastAPI, Header, Query
+
+from app.clients.crud_client import NotificationsCrudClient
+from app.domain.models import CustomerNotificationPreferencesResponse
+from app.services.notification_preferences_service import NotificationPreferencesService
+from app.settings import settings
+
+app = FastAPI(title="SE Notifications example")
+
+
+def get_service() -> NotificationPreferencesService:
+    return NotificationPreferencesService(NotificationsCrudClient(settings.notifications_crud_base_url, settings.customers_crud_base_url))
+
+
+@app.get("/customers/notificationpreferences", response_model=CustomerNotificationPreferencesResponse)
+async def get_customer_notification_preferences(
+    x_htm_customer_profile_id_header: int = Header(..., alias="X-HTM-CUSTOMER-PROFILE-ID-HEADER"),
+    device_id: str | None = Query(default=None, alias="deviceId"),
+):
+    return await get_service().get_customer_notification_preferences(
+        customer_profile_id=x_htm_customer_profile_id_header,
+        device_id=device_id,
+    )

src/python/notificatiesbeheren/notification-preferences-example/app/services/notification_preferences_service.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+from collections import defaultdict
+from typing import Any, Optional
+
+from app.clients.crud_client import NotificationsCrudClient
+from app.domain.models import (
+    CustomerNotificationPreferencesResponse,
+    EventTypeChannelPreference,
+    EventTypePreference,
+    NotificationCategoryPreference,
+    NotificationSubscriptionPreference,
+    OvPayToken,
+    Resource,
+    latest_activity_is_active,
+)
+
+RESOURCE_DEVICES_ID = 8
+RESOURCE_PERSONS_ID = 4
+
+
+class NotificationPreferencesService:
+    def __init__(self, crud_client: NotificationsCrudClient):
+        self.crud_client = crud_client
+
+    async def get_customer_notification_preferences(
+        self,
+        customer_profile_id: int,
+        device_id: Optional[str] = None,
+    ) -> CustomerNotificationPreferencesResponse:
+        subscriptions = await self.crud_client.get_notification_subscriptions(customer_profile_id)
+        categories_by_id: dict[int, NotificationCategoryPreference] = {}
+
+        # Small in-request caches avoid repeated calls for the same category/resources.
+        default_category_cache: dict[int, dict[str, Any]] = {}
+        resource_cache: dict[int, list[dict[str, Any]]] = {}
+
+        for subscription in subscriptions:
+            category_id = _extract_category_id(subscription)
+            default_category = default_category_cache.get(category_id)
+            if default_category is None:
+                default_category = await self.crud_client.get_notification_category_defaults(category_id)
+                if default_category is None:
+                    # Defensive choice: skip orphaned subscriptions rather than returning incomplete UI data.
+                    continue
+                default_category_cache[category_id] = default_category
+
+            category_response = categories_by_id.setdefault(
+                category_id,
+                NotificationCategoryPreference(
+                    notificationCategoryId=category_id,
+                    name=default_category.get("name") or subscription.get("notificationCategory", {}).get("name"),
+                    groupName=default_category.get("groupName") or subscription.get("notificationCategory", {}).get("groupName"),
+                ),
+            )
+
+            token_payload = await self.crud_client.get_ovpay_token(subscription.get("ovPayTokenId"))
+            token = _to_ovpay_token(token_payload, subscription.get("ovPayTokenId"))
+
+            subscription_response = NotificationSubscriptionPreference(
+                notificationSubscriptionId=subscription["notificationSubscriptionId"],
+                customerProfileId=subscription["customerProfileId"],
+                ovPayToken=token,
+                isActive=latest_activity_is_active(subscription),
+            )
+
+            preferences_by_event_type_channel = _index_preferences(subscription)
+
+            for event_type in default_category.get("eventTypes", []):
+                event_type_response = EventTypePreference(
+                    eventTypeId=event_type["eventTypeId"],
+                    name=event_type["name"],
+                    subName=event_type.get("subName"),
+                    prettyName=event_type.get("prettyName") or event_type["name"],
+                )
+
+                for event_type_channel in event_type.get("eventTypeChannels", []):
+                    channel = event_type_channel.get("channel", {})
+                    resource_name = channel.get("resourceName", {})
+                    resource_name_id = resource_name.get("resourceNameId")
+                    resource_name_text = resource_name.get("name") or channel.get("name")
+
+                    if resource_name_id not in resource_cache:
+                        resource_cache[resource_name_id] = await self._load_resources(resource_name_id, customer_profile_id)
+                    source_resources = resource_cache[resource_name_id]
+
+                    if device_id and resource_name_id == RESOURCE_DEVICES_ID:
+                        source_resources = [r for r in source_resources if str(_resource_identifier(r)) == str(device_id)]
+                        if not source_resources:
+                            # SE swagger states a deviceId filter returns only ETCs with resources for that device.
+                            continue
+
+                    matching_preferences = preferences_by_event_type_channel.get(
+                        str(event_type_channel["eventTypeChannelId"]), []
+                    )
+
+                    resources = self._normalize_resources(
+                        resource_name=resource_name_text,
+                        event_type_channel=event_type_channel,
+                        source_resources=source_resources,
+                        preferences=matching_preferences,
+                    )
+
+                    event_type_response.eventTypeChannels.append(
+                        EventTypeChannelPreference(
+                            eventTypeChannelId=event_type_channel["eventTypeChannelId"],
+                            channelId=channel["channelId"],
+                            name=channel["name"],
+                            isMandatory=bool(event_type_channel.get("isMandatory")),
+                            resources=resources,
+                        )
+                    )
+
+                if event_type_response.eventTypeChannels:
+                    subscription_response.eventTypes.append(event_type_response)
+
+            category_response.notificationSubscriptions.append(subscription_response)
+
+        return CustomerNotificationPreferencesResponse(notificationCategories=list(categories_by_id.values()))
+
+    async def _load_resources(self, resource_name_id: int, customer_profile_id: int) -> list[dict[str, Any]]:
+        if resource_name_id == RESOURCE_DEVICES_ID:
+            return await self.crud_client.get_devices(customer_profile_id)
+        if resource_name_id == RESOURCE_PERSONS_ID:
+            return await self.crud_client.get_persons(customer_profile_id)
+        return []
+
+    def _normalize_resources(
+        self,
+        resource_name: str,
+        event_type_channel: dict[str, Any],
+        source_resources: list[dict[str, Any]],
+        preferences: list[dict[str, Any]],
+    ) -> list[Resource]:
+        default_active = bool(event_type_channel.get("isDefault"))
+
+        if not preferences:
+            return [
+                Resource(
+                    resourceName=resource_name,
+                    resourceIdentifier=str(_resource_identifier(resource)),
+                    alias=_resource_alias(resource),
+                    isActive=default_active,
+                )
+                for resource in source_resources
+            ]
+
+        # Preference rows are exceptions/deviations. A null resourceIdentifier means the exception applies to all resources.
+        preference_by_resource: dict[str, dict[str, Any]] = {}
+        global_preference: Optional[dict[str, Any]] = None
+        for preference in preferences:
+            pref_resource_id = preference.get("resourceIdentifier")
+            if pref_resource_id is None:
+                global_preference = preference
+            else:
+                preference_by_resource[str(pref_resource_id)] = preference
+
+        normalized: list[Resource] = []
+        for resource in source_resources:
+            resource_id = str(_resource_identifier(resource))
+            preference = preference_by_resource.get(resource_id) or global_preference
+            if preference is None:
+                is_active = default_active
+            else:
+                # The CRUD example only shows the existence of a preference row, no explicit isActive.
+                # Therefore: for default-on channels a row means opted out; for default-off a row means opted in.
+                is_active = bool(preference.get("isActive", not default_active))
+
+            normalized.append(
+                Resource(
+                    resourceName=resource_name,
+                    resourceIdentifier=resource_id,
+                    alias=_resource_alias(resource),
+                    isActive=is_active,
+                )
+            )
+        return normalized
+
+
+def _extract_category_id(subscription: dict[str, Any]) -> int:
+    if subscription.get("notificationCategoryId") is not None:
+        return int(subscription["notificationCategoryId"])
+    return int(subscription["notificationCategory"]["notificationCategoryId"])
+
+
+def _index_preferences(subscription: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
+    result: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for preference in subscription.get("notificationPreferences", []) or []:
+        event_type_channel_id = preference.get("eventTypeChannelId")
+        if isinstance(event_type_channel_id, dict):
+            event_type_channel_id = event_type_channel_id.get("eventTypeChannelId")
+        if event_type_channel_id is None and isinstance(preference.get("eventTypeChannel"), dict):
+            event_type_channel_id = preference["eventTypeChannel"].get("eventTypeChannelId")
+        if event_type_channel_id is not None:
+            result[str(event_type_channel_id)].append(preference)
+    return result
+
+
+def _to_ovpay_token(payload: Optional[dict[str, Any]], fallback_id: Optional[int]) -> Optional[OvPayToken]:
+    if fallback_id is None:
+        return None
+    # Support both {ovPayToken: {...}} and direct {...} shapes.
+    token = (payload or {}).get("ovPayToken", payload or {})
+    return OvPayToken(ovPayTokenId=token.get("ovPayTokenId", fallback_id), alias=token.get("alias"))
+
+
+def _resource_identifier(resource: dict[str, Any]) -> Any:
+    # Device resources use deviceId. Customer/person email resources are normalized
+    # to customerProfileId because customers-crud-v2 exposes person under /customers.
+    return (
+        resource.get("resourceIdentifier")
+        or resource.get("deviceId")
+        or resource.get("personId")
+        or resource.get("customerProfileId")
+        or resource.get("id")
+    )
+
+
+def _resource_alias(resource: dict[str, Any]) -> Optional[str]:
+    person = resource.get("person") or {}
+    return (
+        resource.get("alias")
+        or resource.get("emailAddress")
+        or resource.get("emailAddresses")
+        or resource.get("email")
+        or person.get("emailAddress")
+        or person.get("emailAddresses")
+        or resource.get("name")
+    )

src/python/notificatiesbeheren/notification-preferences-example/app/settings.py

@@ -0,0 +1,9 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    notifications_crud_base_url: str = "http://localhost:8001"
+    customers_crud_base_url: str = "http://localhost:8002"
+
+
+settings = Settings()

src/python/notificatiesbeheren/notification-preferences-example/requirements.txt

@@ -0,0 +1,8 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+httpx==0.28.1
+pydantic==2.10.4
+pydantic-settings==2.7.1
+azure-functions==1.21.3
+pytest==8.3.4
+pytest-asyncio==0.25.2

src/python/notificatiesbeheren/notification-preferences-example/tests/test_transformation.py

@@ -0,0 +1,83 @@
+import pytest
+
+from app.services.notification_preferences_service import NotificationPreferencesService
+
+
+class FakeCrudClient:
+    async def get_notification_subscriptions(self, customer_profile_id: int):
+        return [
+            {
+                "notificationSubscriptionId": "sub-1",
+                "notificationCategory": {"notificationCategoryId": 1, "name": "Reizen", "groupName": "Mijn passen"},
+                "customerProfileId": customer_profile_id,
+                "ovPayTokenId": 112,
+                "subscriptionActivities": [{"timestamp": "2026-01-01T00:00:00Z", "isActive": True}],
+                "notificationPreferences": [
+                    # Default push=true, this row switches device-2 off only.
+                    {"notificationPreferenceId": "pref-1", "eventTypeChannelId": "etc-push", "resourceIdentifier": "device-2"},
+                    # Default email=false, this row switches customer email on.
+                    {"notificationPreferenceId": "pref-2", "eventTypeChannelId": "etc-email", "resourceIdentifier": "42"},
+                ],
+            }
+        ]
+
+    async def get_notification_category_defaults(self, notification_category_id: int):
+        return {
+            "notificationCategoryId": 1,
+            "name": "Reizen",
+            "groupName": "Mijn passen",
+            "eventTypes": [
+                {
+                    "eventTypeId": 2,
+                    "name": "GBO",
+                    "subName": "CI",
+                    "prettyName": "Normal Check-in",
+                    "eventTypeChannels": [
+                        {
+                            "eventTypeChannelId": "etc-push",
+                            "channel": {"channelId": 1, "name": "push", "resourceName": {"resourceNameId": 8, "name": "devices"}},
+                            "isDefault": True,
+                            "isMandatory": False,
+                        },
+                        {
+                            "eventTypeChannelId": "etc-email",
+                            "channel": {"channelId": 2, "name": "email", "resourceName": {"resourceNameId": 4, "name": "customers"}},
+                            "isDefault": False,
+                            "isMandatory": False,
+                        },
+                    ],
+                }
+            ],
+        }
+
+    async def get_ovpay_token(self, ovpay_token_id):
+        return {"ovPayToken": {"ovPayTokenId": ovpay_token_id, "alias": "Mijn ING bankpas"}}
+
+    async def get_devices(self, customer_profile_id: int):
+        return [
+            {"deviceId": "device-1", "alias": "Mijn iPhone"},
+            {"deviceId": "device-2", "alias": "Mijn Pixel"},
+        ]
+
+    async def get_persons(self, customer_profile_id: int):
+        return [{"customerProfileId": 42, "emailAddress": "customer@example.nl"}]
+
+
+@pytest.mark.asyncio
+async def test_normalizes_defaults_and_customer_exceptions():
+    result = await NotificationPreferencesService(FakeCrudClient()).get_customer_notification_preferences(42)
+    event_type = result.notificationCategories[0].notificationSubscriptions[0].eventTypes[0]
+
+    push_resources = event_type.eventTypeChannels[0].resources
+    assert [r.isActive for r in push_resources] == [True, False]
+
+    email_resources = event_type.eventTypeChannels[1].resources
+    assert email_resources[0].isActive is True
+
+
+@pytest.mark.asyncio
+async def test_device_filter_only_returns_that_device_resource():
+    result = await NotificationPreferencesService(FakeCrudClient()).get_customer_notification_preferences(42, device_id="device-1")
+    event_type = result.notificationCategories[0].notificationSubscriptions[0].eventTypes[0]
+    assert len(event_type.eventTypeChannels[0].resources) == 1
+    assert event_type.eventTypeChannels[0].resources[0].resourceIdentifier == "device-1"