feat(ticketing): add dynamic content items and variants

libzapi/application/commands/ticketing/dynamic_content_cmds.py

@@ -0,0 +1,38 @@
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+
+@dataclass(frozen=True, slots=True)
+class DynamicContentVariantInputCmd:
+    content: str
+    locale_id: int
+    default: Optional[bool] = None
+    active: Optional[bool] = None
+
+
+@dataclass(frozen=True, slots=True)
+class CreateDynamicContentItemCmd:
+    name: str
+    default_locale_id: int
+    variants: List[DynamicContentVariantInputCmd] = field(default_factory=list)
+
+
+@dataclass(frozen=True, slots=True)
+class UpdateDynamicContentItemCmd:
+    name: Optional[str] = None
+
+
+@dataclass(frozen=True, slots=True)
+class CreateDynamicContentVariantCmd:
+    content: str
+    locale_id: int
+    default: Optional[bool] = None
+    active: Optional[bool] = None
+
+
+@dataclass(frozen=True, slots=True)
+class UpdateDynamicContentVariantCmd:
+    content: Optional[str] = None
+    locale_id: Optional[int] = None
+    default: Optional[bool] = None
+    active: Optional[bool] = None

libzapi/application/services/ticketing/__init__.py

@@ -7,6 +7,7 @@
 from libzapi.application.services.ticketing.custom_ticket_statuses_service import (
     CustomTicketStatusesService,
 )
+from libzapi.application.services.ticketing.dynamic_content_service import DynamicContentService
 from libzapi.application.services.ticketing.email_notifications_service import EmailNotificationService
 from libzapi.application.services.ticketing.groups_service import GroupsService
 from libzapi.application.services.ticketing.group_memberships_service import (
@@ -79,6 +80,7 @@ def __init__(
         self.custom_ticket_statuses = CustomTicketStatusesService(
             api.CustomTicketStatusApiClient(http)
         )
+        self.dynamic_content = DynamicContentService(api.DynamicContentApiClient(http))
         self.email_notifications = EmailNotificationService(api.EmailNotificationApiClient(http))
         self.groups = GroupsService(api.GroupApiClient(http))
         self.group_memberships = GroupMembershipsService(api.GroupMembershipApiClient(http))

libzapi/application/services/ticketing/dynamic_content_service.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from typing import Iterator, List, Optional
+
+from libzapi.application.commands.ticketing.dynamic_content_cmds import (
+    CreateDynamicContentItemCmd,
+    CreateDynamicContentVariantCmd,
+    DynamicContentVariantInputCmd,
+    UpdateDynamicContentItemCmd,
+    UpdateDynamicContentVariantCmd,
+)
+from libzapi.domain.models.ticketing.dynamic_content import (
+    DynamicContentItem,
+    DynamicContentVariant,
+)
+from libzapi.infrastructure.api_clients.ticketing.dynamic_content_api_client import (
+    DynamicContentApiClient,
+)
+
+
+class DynamicContentService:
+    def __init__(self, client: DynamicContentApiClient) -> None:
+        self._client = client
+
+    def list_items(self) -> Iterator[DynamicContentItem]:
+        return self._client.list_items()
+
+    def get_item(self, item_id: int) -> DynamicContentItem:
+        return self._client.get_item(item_id=item_id)
+
+    def create_item(
+        self,
+        *,
+        name: str,
+        default_locale_id: int,
+        variants: Optional[List[DynamicContentVariantInputCmd]] = None,
+    ) -> DynamicContentItem:
+        cmd = CreateDynamicContentItemCmd(
+            name=name,
+            default_locale_id=default_locale_id,
+            variants=list(variants) if variants else [],
+        )
+        return self._client.create_item(entity=cmd)
+
+    def update_item(
+        self, item_id: int, *, name: Optional[str] = None
+    ) -> DynamicContentItem:
+        cmd = UpdateDynamicContentItemCmd(name=name)
+        return self._client.update_item(item_id=item_id, entity=cmd)
+
+    def delete_item(self, item_id: int) -> None:
+        self._client.delete_item(item_id=item_id)
+
+    def list_variants(self, item_id: int) -> Iterator[DynamicContentVariant]:
+        return self._client.list_variants(item_id=item_id)
+
+    def get_variant(
+        self, item_id: int, variant_id: int
+    ) -> DynamicContentVariant:
+        return self._client.get_variant(item_id=item_id, variant_id=variant_id)
+
+    def create_variant(
+        self,
+        item_id: int,
+        *,
+        content: str,
+        locale_id: int,
+        default: Optional[bool] = None,
+        active: Optional[bool] = None,
+    ) -> DynamicContentVariant:
+        cmd = CreateDynamicContentVariantCmd(
+            content=content,
+            locale_id=locale_id,
+            default=default,
+            active=active,
+        )
+        return self._client.create_variant(item_id=item_id, entity=cmd)
+
+    def update_variant(
+        self,
+        item_id: int,
+        variant_id: int,
+        *,
+        content: Optional[str] = None,
+        locale_id: Optional[int] = None,
+        default: Optional[bool] = None,
+        active: Optional[bool] = None,
+    ) -> DynamicContentVariant:
+        cmd = UpdateDynamicContentVariantCmd(
+            content=content,
+            locale_id=locale_id,
+            default=default,
+            active=active,
+        )
+        return self._client.update_variant(
+            item_id=item_id, variant_id=variant_id, entity=cmd
+        )
+
+    def delete_variant(self, item_id: int, variant_id: int) -> None:
+        self._client.delete_variant(item_id=item_id, variant_id=variant_id)

libzapi/domain/models/ticketing/dynamic_content.py

@@ -0,0 +1,39 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import List, Optional
+
+from libzapi.domain.shared_objects.logical_key import LogicalKey
+
+
+@dataclass(frozen=True, slots=True)
+class DynamicContentVariant:
+    id: int
+    content: str
+    locale_id: int
+    default: bool = False
+    active: bool = True
+    outdated: bool = False
+    url: Optional[str] = None
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+    @property
+    def logical_key(self) -> LogicalKey:
+        return LogicalKey("dynamic_content_variant", f"variant_id_{self.id}")
+
+
+@dataclass(frozen=True, slots=True)
+class DynamicContentItem:
+    id: int
+    name: str
+    placeholder: str
+    default_locale_id: int
+    url: Optional[str] = None
+    outdated: bool = False
+    variants: List[DynamicContentVariant] = field(default_factory=list)
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+    @property
+    def logical_key(self) -> LogicalKey:
+        return LogicalKey("dynamic_content_item", f"item_id_{self.id}")

libzapi/infrastructure/api_clients/ticketing/__init__.py

@@ -2,6 +2,9 @@
 from libzapi.infrastructure.api_clients.ticketing.attachment_api_client import AttachmentApiClient
 from libzapi.infrastructure.api_clients.ticketing.automation_api_client import AutomationApiClient
 from libzapi.infrastructure.api_clients.ticketing.brand_api_client import BrandApiClient
+from libzapi.infrastructure.api_clients.ticketing.dynamic_content_api_client import (
+    DynamicContentApiClient,
+)
 from libzapi.infrastructure.api_clients.ticketing.brand_agent_api_client import BrandAgentApiClient
 from libzapi.infrastructure.api_clients.ticketing.custom_ticket_status_api_client import (
     CustomTicketStatusApiClient,
@@ -61,6 +64,7 @@
     "BrandApiClient",
     "BrandAgentApiClient",
     "CustomTicketStatusApiClient",
+    "DynamicContentApiClient",
     "EmailNotificationApiClient",
     "GroupApiClient",
     "GroupMembershipApiClient",

libzapi/infrastructure/api_clients/ticketing/dynamic_content_api_client.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+from typing import Iterator
+
+from libzapi.application.commands.ticketing.dynamic_content_cmds import (
+    CreateDynamicContentItemCmd,
+    CreateDynamicContentVariantCmd,
+    UpdateDynamicContentItemCmd,
+    UpdateDynamicContentVariantCmd,
+)
+from libzapi.domain.models.ticketing.dynamic_content import (
+    DynamicContentItem,
+    DynamicContentVariant,
+)
+from libzapi.infrastructure.http.client import HttpClient
+from libzapi.infrastructure.http.pagination import yield_items
+from libzapi.infrastructure.mappers.ticketing.dynamic_content_mapper import (
+    to_payload_create_item,
+    to_payload_create_variant,
+    to_payload_update_item,
+    to_payload_update_variant,
+)
+from libzapi.infrastructure.serialization.parse import to_domain
+
+_ITEMS = "/api/v2/dynamic_content/items"
+
+
+class DynamicContentApiClient:
+    """HTTP adapter for Zendesk Dynamic Content."""
+
+    def __init__(self, http: HttpClient) -> None:
+        self._http = http
+
+    def list_items(self) -> Iterator[DynamicContentItem]:
+        for obj in yield_items(
+            get_json=self._http.get,
+            first_path=_ITEMS,
+            base_url=self._http.base_url,
+            items_key="items",
+        ):
+            yield to_domain(data=obj, cls=DynamicContentItem)
+
+    def get_item(self, item_id: int) -> DynamicContentItem:
+        data = self._http.get(f"{_ITEMS}/{int(item_id)}")
+        return to_domain(data=data["item"], cls=DynamicContentItem)
+
+    def create_item(
+        self, entity: CreateDynamicContentItemCmd
+    ) -> DynamicContentItem:
+        data = self._http.post(_ITEMS, to_payload_create_item(entity))
+        return to_domain(data=data["item"], cls=DynamicContentItem)
+
+    def update_item(
+        self, item_id: int, entity: UpdateDynamicContentItemCmd
+    ) -> DynamicContentItem:
+        data = self._http.put(
+            f"{_ITEMS}/{int(item_id)}", to_payload_update_item(entity)
+        )
+        return to_domain(data=data["item"], cls=DynamicContentItem)
+
+    def delete_item(self, item_id: int) -> None:
+        self._http.delete(f"{_ITEMS}/{int(item_id)}")
+
+    def list_variants(self, item_id: int) -> Iterator[DynamicContentVariant]:
+        path = f"{_ITEMS}/{int(item_id)}/variants"
+        for obj in yield_items(
+            get_json=self._http.get,
+            first_path=path,
+            base_url=self._http.base_url,
+            items_key="variants",
+        ):
+            yield to_domain(data=obj, cls=DynamicContentVariant)
+
+    def get_variant(self, item_id: int, variant_id: int) -> DynamicContentVariant:
+        data = self._http.get(
+            f"{_ITEMS}/{int(item_id)}/variants/{int(variant_id)}"
+        )
+        return to_domain(data=data["variant"], cls=DynamicContentVariant)
+
+    def create_variant(
+        self, item_id: int, entity: CreateDynamicContentVariantCmd
+    ) -> DynamicContentVariant:
+        data = self._http.post(
+            f"{_ITEMS}/{int(item_id)}/variants",
+            to_payload_create_variant(entity),
+        )
+        return to_domain(data=data["variant"], cls=DynamicContentVariant)
+
+    def update_variant(
+        self,
+        item_id: int,
+        variant_id: int,
+        entity: UpdateDynamicContentVariantCmd,
+    ) -> DynamicContentVariant:
+        data = self._http.put(
+            f"{_ITEMS}/{int(item_id)}/variants/{int(variant_id)}",
+            to_payload_update_variant(entity),
+        )
+        return to_domain(data=data["variant"], cls=DynamicContentVariant)
+
+    def delete_variant(self, item_id: int, variant_id: int) -> None:
+        self._http.delete(f"{_ITEMS}/{int(item_id)}/variants/{int(variant_id)}")

libzapi/infrastructure/mappers/ticketing/dynamic_content_mapper.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from libzapi.application.commands.ticketing.dynamic_content_cmds import (
+    CreateDynamicContentItemCmd,
+    CreateDynamicContentVariantCmd,
+    DynamicContentVariantInputCmd,
+    UpdateDynamicContentItemCmd,
+    UpdateDynamicContentVariantCmd,
+)
+
+
+def _variant_input_to_payload(variant: DynamicContentVariantInputCmd) -> dict:
+    payload: dict = {
+        "content": variant.content,
+        "locale_id": int(variant.locale_id),
+    }
+    if variant.default is not None:
+        payload["default"] = variant.default
+    if variant.active is not None:
+        payload["active"] = variant.active
+    return payload
+
+
+def to_payload_create_item(cmd: CreateDynamicContentItemCmd) -> dict:
+    return {
+        "item": {
+            "name": cmd.name,
+            "default_locale_id": int(cmd.default_locale_id),
+            "variants": [_variant_input_to_payload(v) for v in cmd.variants],
+        }
+    }
+
+
+def to_payload_update_item(cmd: UpdateDynamicContentItemCmd) -> dict:
+    inner: dict = {}
+    if cmd.name is not None:
+        inner["name"] = cmd.name
+    return {"item": inner}
+
+
+def to_payload_create_variant(cmd: CreateDynamicContentVariantCmd) -> dict:
+    inner: dict = {
+        "content": cmd.content,
+        "locale_id": int(cmd.locale_id),
+    }
+    if cmd.default is not None:
+        inner["default"] = cmd.default
+    if cmd.active is not None:
+        inner["active"] = cmd.active
+    return {"variant": inner}
+
+
+def to_payload_update_variant(cmd: UpdateDynamicContentVariantCmd) -> dict:
+    inner: dict = {}
+    if cmd.content is not None:
+        inner["content"] = cmd.content
+    if cmd.locale_id is not None:
+        inner["locale_id"] = int(cmd.locale_id)
+    if cmd.default is not None:
+        inner["default"] = cmd.default
+    if cmd.active is not None:
+        inner["active"] = cmd.active
+    return {"variant": inner}