feat(ticketing): add views CUD, search, count, execute

libzapi/application/commands/ticketing/view_cmds.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Iterable, TypeAlias
+
+
+@dataclass(frozen=True, slots=True)
+class CreateViewCmd:
+    title: str
+    all: Iterable[dict[str, Any]] | None = None
+    any: Iterable[dict[str, Any]] | None = None
+    description: str | None = None
+    active: bool | None = None
+    position: int | None = None
+    output: dict[str, Any] | None = None
+    restriction: dict[str, Any] | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class UpdateViewCmd:
+    title: str | None = None
+    all: Iterable[dict[str, Any]] | None = None
+    any: Iterable[dict[str, Any]] | None = None
+    description: str | None = None
+    active: bool | None = None
+    position: int | None = None
+    output: dict[str, Any] | None = None
+    restriction: dict[str, Any] | None = None
+
+
+ViewCmd: TypeAlias = CreateViewCmd | UpdateViewCmd

libzapi/application/services/ticketing/views_service.py

@@ -1,7 +1,14 @@
-from typing import Iterable
+from __future__ import annotations
 
+from typing import Any, Iterable
+
+from libzapi.application.commands.ticketing.view_cmds import (
+    CreateViewCmd,
+    UpdateViewCmd,
+)
 from libzapi.domain.models.ticketing.view import View
 from libzapi.domain.shared_objects.count_snapshot import CountSnapshot
+from libzapi.domain.shared_objects.job_status import JobStatus
 from libzapi.infrastructure.api_clients.ticketing.view_api_client import ViewApiClient
 
 
@@ -14,11 +21,46 @@ def __init__(self, client: ViewApiClient) -> None:
     def list_all(self) -> Iterable[View]:
         return self._client.list_all()
 
-    def list_active(self, view_id: int) -> View:
-        return self._client.get(view_id=view_id)
+    def list_active(self) -> Iterable[View]:
+        return self._client.list_active()
+
+    def search(self, query: str) -> Iterable[View]:
+        return self._client.search(query=query)
 
     def count(self) -> CountSnapshot:
         return self._client.count()
 
+    def count_view(self, view_id: int) -> dict:
+        return self._client.count_view(view_id=view_id)
+
+    def count_many(self, view_ids: Iterable[int]) -> list[dict]:
+        return self._client.count_many(view_ids=view_ids)
+
+    def execute(self, view_id: int) -> dict:
+        return self._client.execute(view_id=view_id)
+
     def get_by_id(self, view_id: int) -> View:
-        return self._client.get(view_id)
+        return self._client.get(view_id=view_id)
+
+    def create(self, **fields) -> View:
+        return self._client.create(entity=CreateViewCmd(**fields))
+
+    def update(self, view_id: int, **fields) -> View:
+        return self._client.update(
+            view_id=view_id, entity=UpdateViewCmd(**fields)
+        )
+
+    def delete(self, view_id: int) -> None:
+        self._client.delete(view_id=view_id)
+
+    def update_many(
+        self, updates: Iterable[tuple[int, dict[str, Any]]]
+    ) -> JobStatus:
+        pairs = [
+            (view_id, UpdateViewCmd(**fields))
+            for view_id, fields in updates
+        ]
+        return self._client.update_many(updates=pairs)
+
+    def destroy_many(self, view_ids: Iterable[int]) -> JobStatus:
+        return self._client.destroy_many(view_ids=view_ids)

libzapi/infrastructure/api_clients/ticketing/view_api_client.py

@@ -1,13 +1,22 @@
 from __future__ import annotations
-from typing import Iterable
 
+from typing import Iterable, Iterator
 
+from libzapi.application.commands.ticketing.view_cmds import (
+    CreateViewCmd,
+    UpdateViewCmd,
+)
+from libzapi.domain.models.ticketing.view import View
 from libzapi.domain.shared_objects.count_snapshot import CountSnapshot
-from libzapi.infrastructure.mappers.count_mapper import to_count_snapshot
+from libzapi.domain.shared_objects.job_status import JobStatus
 from libzapi.infrastructure.http.client import HttpClient
 from libzapi.infrastructure.http.pagination import yield_items
+from libzapi.infrastructure.mappers.count_mapper import to_count_snapshot
+from libzapi.infrastructure.mappers.ticketing.view_mapper import (
+    to_payload_create,
+    to_payload_update,
+)
 from libzapi.infrastructure.serialization.parse import to_domain
-from libzapi.domain.models.ticketing.view import View
 
 
 class ViewApiClient:
@@ -16,7 +25,7 @@ class ViewApiClient:
     def __init__(self, http: HttpClient) -> None:
         self._http = http
 
-    def list_all(self) -> Iterable[View]:
+    def list_all(self) -> Iterator[View]:
         for obj in yield_items(
             get_json=self._http.get,
             first_path="/api/v2/views",
@@ -25,7 +34,7 @@ def list_all(self) -> Iterable[View]:
         ):
             yield to_domain(data=obj, cls=View)
 
-    def list_active(self) -> Iterable[View]:
+    def list_active(self) -> Iterator[View]:
         for obj in yield_items(
             get_json=self._http.get,
             first_path="/api/v2/views/active",
@@ -34,10 +43,65 @@ def list_active(self) -> Iterable[View]:
         ):
             yield to_domain(data=obj, cls=View)
 
+    def search(self, query: str) -> Iterator[View]:
+        for obj in yield_items(
+            get_json=self._http.get,
+            first_path=f"/api/v2/views/search?query={query}",
+            base_url=self._http.base_url,
+            items_key="views",
+        ):
+            yield to_domain(data=obj, cls=View)
+
     def count(self) -> CountSnapshot:
         data = self._http.get("/api/v2/views/count")
         return to_count_snapshot(data["count"])
 
+    def count_view(self, view_id: int) -> dict:
+        data = self._http.get(f"/api/v2/views/{int(view_id)}/count")
+        return data.get("view_count", {})
+
+    def count_many(self, view_ids: Iterable[int]) -> list[dict]:
+        ids_str = ",".join(str(int(i)) for i in view_ids)
+        data = self._http.get(f"/api/v2/views/count_many?ids={ids_str}")
+        return list(data.get("view_counts", []))
+
+    def execute(self, view_id: int) -> dict:
+        return self._http.get(f"/api/v2/views/{int(view_id)}/execute")
+
     def get(self, view_id: int) -> View:
         data = self._http.get(f"/api/v2/views/{int(view_id)}")
         return to_domain(data=data["view"], cls=View)
+
+    def create(self, entity: CreateViewCmd) -> View:
+        payload = to_payload_create(entity)
+        data = self._http.post("/api/v2/views", payload)
+        return to_domain(data=data["view"], cls=View)
+
+    def update(self, view_id: int, entity: UpdateViewCmd) -> View:
+        payload = to_payload_update(entity)
+        data = self._http.put(f"/api/v2/views/{int(view_id)}", payload)
+        return to_domain(data=data["view"], cls=View)
+
+    def delete(self, view_id: int) -> None:
+        self._http.delete(f"/api/v2/views/{int(view_id)}")
+
+    def update_many(
+        self, updates: Iterable[tuple[int, UpdateViewCmd]]
+    ) -> JobStatus:
+        items = []
+        for view_id, cmd in updates:
+            item = to_payload_update(cmd)["view"]
+            item["id"] = int(view_id)
+            items.append(item)
+        data = self._http.put(
+            "/api/v2/views/update_many", {"views": items}
+        )
+        return to_domain(data=data["job_status"], cls=JobStatus)
+
+    def destroy_many(self, view_ids: Iterable[int]) -> JobStatus:
+        ids_str = ",".join(str(int(i)) for i in view_ids)
+        data = (
+            self._http.delete(f"/api/v2/views/destroy_many?ids={ids_str}")
+            or {}
+        )
+        return to_domain(data=data["job_status"], cls=JobStatus)

libzapi/infrastructure/mappers/ticketing/view_mapper.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from libzapi.application.commands.ticketing.view_cmds import (
+    CreateViewCmd,
+    UpdateViewCmd,
+)
+
+
+def to_payload_create(cmd: CreateViewCmd) -> dict:
+    body: dict = {"title": cmd.title}
+    if cmd.all is not None:
+        body["all"] = list(cmd.all)
+    if cmd.any is not None:
+        body["any"] = list(cmd.any)
+    if cmd.description is not None:
+        body["description"] = cmd.description
+    if cmd.active is not None:
+        body["active"] = cmd.active
+    if cmd.position is not None:
+        body["position"] = cmd.position
+    if cmd.output is not None:
+        body["output"] = cmd.output
+    if cmd.restriction is not None:
+        body["restriction"] = cmd.restriction
+    return {"view": body}
+
+
+def to_payload_update(cmd: UpdateViewCmd) -> dict:
+    body: dict = {}
+    if cmd.title is not None:
+        body["title"] = cmd.title
+    if cmd.all is not None:
+        body["all"] = list(cmd.all)
+    if cmd.any is not None:
+        body["any"] = list(cmd.any)
+    if cmd.description is not None:
+        body["description"] = cmd.description
+    if cmd.active is not None:
+        body["active"] = cmd.active
+    if cmd.position is not None:
+        body["position"] = cmd.position
+    if cmd.output is not None:
+        body["output"] = cmd.output
+    if cmd.restriction is not None:
+        body["restriction"] = cmd.restriction
+    return {"view": body}

tests/integration/ticketing/test_view.py

@@ -0,0 +1,97 @@
+import itertools
+import uuid
+
+from libzapi import Ticketing
+
+
+def _unique() -> str:
+    return uuid.uuid4().hex[:10]
+
+
+def _create_view(ticketing: Ticketing, **overrides):
+    suffix = _unique()
+    defaults = dict(
+        title=f"libzapi view {suffix}",
+        all=[{"field": "status", "operator": "is", "value": "open"}],
+        output={"columns": ["subject", "requester", "created"]},
+    )
+    defaults.update(overrides)
+    return ticketing.views.create(**defaults)
+
+
+def test_list_and_get_view(ticketing: Ticketing):
+    views = list(itertools.islice(ticketing.views.list_all(), 20))
+    assert len(views) > 0
+    view = ticketing.views.get_by_id(views[0].id)
+    assert view.raw_title == views[0].raw_title
+
+
+def test_list_active(ticketing: Ticketing):
+    views = list(itertools.islice(ticketing.views.list_active(), 20))
+    assert isinstance(views, list)
+
+
+def test_count(ticketing: Ticketing):
+    snapshot = ticketing.views.count()
+    assert snapshot.value is not None
+
+
+def test_create_update_delete(ticketing: Ticketing):
+    view = _create_view(ticketing, description="created by libzapi")
+    assert view.id > 0
+    updated = ticketing.views.update(view.id, active=False)
+    assert updated.active is False
+    ticketing.views.delete(view.id)
+
+
+def test_count_view_and_execute(ticketing: Ticketing):
+    view = _create_view(ticketing)
+    try:
+        count = ticketing.views.count_view(view.id)
+        assert isinstance(count, dict)
+        result = ticketing.views.execute(view.id)
+        assert isinstance(result, dict)
+    finally:
+        ticketing.views.delete(view.id)
+
+
+def test_count_many(ticketing: Ticketing):
+    a = _create_view(ticketing)
+    b = _create_view(ticketing)
+    try:
+        counts = ticketing.views.count_many([a.id, b.id])
+        assert isinstance(counts, list)
+    finally:
+        ticketing.views.delete(a.id)
+        ticketing.views.delete(b.id)
+
+
+def test_search(ticketing: Ticketing):
+    view = _create_view(ticketing, title=f"libzapi search {_unique()}")
+    try:
+        matches = list(
+            itertools.islice(ticketing.views.search(query="libzapi"), 10)
+        )
+        assert any(m.id == view.id for m in matches) or matches == []
+    finally:
+        ticketing.views.delete(view.id)
+
+
+def test_update_many(ticketing: Ticketing):
+    a = _create_view(ticketing)
+    b = _create_view(ticketing)
+    try:
+        job = ticketing.views.update_many(
+            [(a.id, {"active": False}), (b.id, {"active": False})]
+        )
+        assert job.id
+    finally:
+        ticketing.views.delete(a.id)
+        ticketing.views.delete(b.id)
+
+
+def test_destroy_many(ticketing: Ticketing):
+    a = _create_view(ticketing)
+    b = _create_view(ticketing)
+    job = ticketing.views.destroy_many([a.id, b.id])
+    assert job.id