[feature]: Add Zendesk Workforce Management (WFM) SDK entry point

libzapi/__init__.py

@@ -4,6 +4,7 @@
 from libzapi.application import AgentAvailability
 from libzapi.application import AssetManagement
 from libzapi.application import ZendeskStatus
+from libzapi.application import WorkforceManagement
 
 __all__ = [
     "HelpCenter",
@@ -12,4 +13,5 @@
     "AgentAvailability",
     "AssetManagement",
     "ZendeskStatus",
+    "WorkforceManagement",
 ]

libzapi/application/__init__.py

@@ -4,6 +4,7 @@
 from libzapi.application.services.agent_availability import AgentAvailability
 from libzapi.application.services.asset_management import AssetManagement
 from libzapi.application.services.status import ZendeskStatus
+from libzapi.application.services.wfm import WorkforceManagement
 
 __all__ = [
     "HelpCenter",
@@ -12,4 +13,5 @@
     "AgentAvailability",
     "AssetManagement",
     "ZendeskStatus",
+    "WorkforceManagement",
 ]

libzapi/application/commands/wfm/shift_cmds.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class FetchShiftsCmd:
+    startDate: str
+    endDate: str
+    agentIds: list[int] | None = None
+    published: int | None = None
+    page: int = 1

libzapi/application/commands/wfm/team_cmds.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class CreateTeamCmd:
+    name: str
+    description: str
+    manager_id: int
+    agents_ids: list[str]
+
+
+@dataclass(frozen=True, slots=True)
+class UpdateTeamCmd:
+    name: str | None = None
+    description: str | None = None
+    manager_id: int | None = None
+    agents_ids: list[str] | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class BulkAgentsCmd:
+    agent_ids: list[str]
+    team_ids: list[str]

libzapi/application/commands/wfm/time_off_cmds.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True, slots=True)
+class ImportTimeOffEntry:
+    agentId: int
+    startTime: int
+    endTime: int
+    reasonId: str
+    id: str | None = None
+    note: str | None = None
+    status: str | None = None
+    timeOffType: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class ImportTimeOffCmd:
+    data: list[ImportTimeOffEntry] = field(default_factory=list)

libzapi/application/services/wfm/__init__.py

@@ -0,0 +1,28 @@
+import libzapi.infrastructure.api_clients.wfm as api
+from libzapi.application.services.wfm.activities_service import ActivitiesService
+from libzapi.application.services.wfm.reports_service import ReportsService
+from libzapi.application.services.wfm.shifts_service import ShiftsService
+from libzapi.application.services.wfm.teams_service import TeamsService
+from libzapi.application.services.wfm.time_off_service import TimeOffService
+from libzapi.infrastructure.http.auth import api_token_headers, oauth_headers
+from libzapi.infrastructure.http.client import HttpClient
+
+
+class WorkforceManagement:
+    def __init__(
+        self, base_url: str, oauth_token: str | None = None, email: str | None = None, api_token: str | None = None
+    ):
+        if oauth_token:
+            headers = oauth_headers(oauth_token)
+        elif email and api_token:
+            headers = api_token_headers(email, api_token)
+        else:
+            raise ValueError("Provide oauth_token or email+api_token")
+
+        http = HttpClient(base_url, headers=headers)
+
+        self.activities = ActivitiesService(api.ActivityApiClient(http))
+        self.reports = ReportsService(api.ReportApiClient(http))
+        self.shifts = ShiftsService(api.ShiftApiClient(http))
+        self.time_off = TimeOffService(api.TimeOffApiClient(http))
+        self.teams = TeamsService(api.TeamApiClient(http))

libzapi/application/services/wfm/activities_service.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+from libzapi.domain.models.wfm.activity import Activity, ActivityTypeRef, AgentRef
+from libzapi.infrastructure.api_clients.wfm import ActivityApiClient
+
+
+class ActivitiesService:
+    def __init__(self, client: ActivityApiClient) -> None:
+        self._client = client
+
+    def list(self, start_time: int) -> Iterable[Activity]:
+        return self._client.list(start_time=start_time)
+
+    def list_with_relationships(self, start_time: int) -> tuple[list[Activity], list[AgentRef], list[ActivityTypeRef]]:
+        return self._client.list_with_relationships(start_time=start_time)

libzapi/application/services/wfm/reports_service.py

@@ -0,0 +1,17 @@
+from typing import Iterable
+
+from libzapi.domain.models.wfm.report import ReportRow
+from libzapi.infrastructure.api_clients.wfm import ReportApiClient
+
+
+class ReportsService:
+    def __init__(self, client: ReportApiClient) -> None:
+        self._client = client
+
+    def get_data(self, template_id: str, start_time: int, end_time: int) -> Iterable[ReportRow]:
+        return self._client.get_data(template_id=template_id, start_time=start_time, end_time=end_time)
+
+    def get_data_with_relationships(self, template_id: str, start_time: int, end_time: int) -> dict:
+        return self._client.get_data_with_relationships(
+            template_id=template_id, start_time=start_time, end_time=end_time
+        )

libzapi/application/services/wfm/shifts_service.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+from libzapi.application.commands.wfm.shift_cmds import FetchShiftsCmd
+from libzapi.domain.models.wfm.shift import Shift
+from libzapi.infrastructure.api_clients.wfm import ShiftApiClient
+
+
+class ShiftsService:
+    def __init__(self, client: ShiftApiClient) -> None:
+        self._client = client
+
+    def fetch(
+        self,
+        start_date: str,
+        end_date: str,
+        agent_ids: list[int] | None = None,
+        published: int | None = None,
+        page: int = 1,
+    ) -> Iterable[Shift]:
+        cmd = FetchShiftsCmd(
+            startDate=start_date,
+            endDate=end_date,
+            agentIds=agent_ids,
+            published=published,
+            page=page,
+        )
+        return self._client.fetch(cmd=cmd)

libzapi/application/services/wfm/teams_service.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+from libzapi.application.commands.wfm.team_cmds import BulkAgentsCmd, CreateTeamCmd, UpdateTeamCmd
+from libzapi.domain.models.wfm.team import BulkAgentsResult, Team
+from libzapi.infrastructure.api_clients.wfm import TeamApiClient
+
+
+class TeamsService:
+    def __init__(self, client: TeamApiClient) -> None:
+        self._client = client
+
+    def list(self, deleted: bool = False) -> Iterable[Team]:
+        return self._client.list(deleted=deleted)
+
+    def get(self, team_id: str) -> Team:
+        return self._client.get(team_id=team_id)
+
+    def create(self, name: str, description: str, manager_id: int, agents_ids: list[str]) -> Team:
+        cmd = CreateTeamCmd(name=name, description=description, manager_id=manager_id, agents_ids=agents_ids)
+        return self._client.create(cmd=cmd)
+
+    def update(self, team_id: str, **kwargs) -> Team:
+        cmd = UpdateTeamCmd(**kwargs)
+        return self._client.update(team_id=team_id, cmd=cmd)
+
+    def delete(self, team_id: str) -> None:
+        self._client.delete(team_id=team_id)
+
+    def restore(self, team_id: str) -> Team:
+        return self._client.restore(team_id=team_id)
+
+    def bulk_add_agents(self, agent_ids: list[str], team_ids: list[str]) -> BulkAgentsResult:
+        cmd = BulkAgentsCmd(agent_ids=agent_ids, team_ids=team_ids)
+        return self._client.bulk_add_agents(cmd=cmd)
+
+    def bulk_remove_agents(self, agent_ids: list[str], team_ids: list[str]) -> BulkAgentsResult:
+        cmd = BulkAgentsCmd(agent_ids=agent_ids, team_ids=team_ids)
+        return self._client.bulk_remove_agents(cmd=cmd)

libzapi/application/services/wfm/time_off_service.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+from libzapi.application.commands.wfm.time_off_cmds import ImportTimeOffCmd, ImportTimeOffEntry
+from libzapi.domain.models.wfm.time_off import TimeOff, TimeOffImportResult
+from libzapi.infrastructure.api_clients.wfm import TimeOffApiClient
+
+
+class TimeOffService:
+    def __init__(self, client: TimeOffApiClient) -> None:
+        self._client = client
+
+    def list(
+        self,
+        time_off_request_id: str | None = None,
+        agent_id: int | None = None,
+        start_time: int | None = None,
+        end_time: int | None = None,
+        status: str | None = None,
+        reason_id: str | None = None,
+        time_off_type: str | None = None,
+        page: int | None = None,
+        per_page: int | None = None,
+    ) -> Iterable[TimeOff]:
+        return self._client.list(
+            time_off_request_id=time_off_request_id,
+            agent_id=agent_id,
+            start_time=start_time,
+            end_time=end_time,
+            status=status,
+            reason_id=reason_id,
+            time_off_type=time_off_type,
+            page=page,
+            per_page=per_page,
+        )
+
+    def import_time_off(self, entries: list[ImportTimeOffEntry]) -> TimeOffImportResult:
+        cmd = ImportTimeOffCmd(data=entries)
+        return self._client.import_time_off(cmd=cmd)

libzapi/domain/models/wfm/activity.py

@@ -0,0 +1,41 @@
+from dataclasses import dataclass, field
+
+from libzapi.domain.shared_objects.logical_key import LogicalKey
+
+
+@dataclass(frozen=True, slots=True)
+class AgentRef:
+    id: int
+    name: str
+    email: str
+    deactivated: bool = False
+    isDeleted: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class ActivityTypeRef:
+    id: str
+    name: str
+    color: str = ""
+    isDeleted: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class Activity:
+    id: str
+    agentId: int
+    startTime: int
+    type: str
+    name: str = ""
+    ticketId: int | None = None
+    endTime: int | None = None
+    duration: int | None = None
+    activityTypeIds: list[str] = field(default_factory=list)
+    eventType: str = ""
+    color: str = ""
+    isPaid: bool = False
+    lockIntervals: str | None = None
+
+    @property
+    def logical_key(self) -> LogicalKey:
+        return LogicalKey("wfm_activity", self.id)

libzapi/domain/models/wfm/report.py

@@ -0,0 +1,27 @@
+from dataclasses import dataclass, field
+
+from libzapi.domain.shared_objects.logical_key import LogicalKey
+
+
+@dataclass(frozen=True, slots=True)
+class Grouping:
+    key: str
+    value: str
+
+
+@dataclass(frozen=True, slots=True)
+class Metric:
+    key: str
+    value: str
+    type: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class ReportRow:
+    groupings: list[Grouping] = field(default_factory=list)
+    metrics: list[Metric] = field(default_factory=list)
+
+    @property
+    def logical_key(self) -> LogicalKey:
+        keys = "_".join(g.value for g in self.groupings)
+        return LogicalKey("wfm_report_row", keys)

libzapi/domain/models/wfm/shift.py

@@ -0,0 +1,31 @@
+from dataclasses import dataclass, field
+
+from libzapi.domain.shared_objects.logical_key import LogicalKey
+
+
+@dataclass(frozen=True, slots=True)
+class ShiftTask:
+    id: str
+    startTime: int
+    endTime: int
+    name: str = ""
+    color: str = ""
+    taskableId: str = ""
+    taskableType: str = ""
+    createdAt: str = ""
+    note: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class Shift:
+    id: str
+    agentId: int
+    startTime: int
+    endTime: int
+    published: bool = False
+    parentId: str | int | None = None
+    tasks: list[ShiftTask] = field(default_factory=list)
+
+    @property
+    def logical_key(self) -> LogicalKey:
+        return LogicalKey("wfm_shift", self.id)

libzapi/domain/models/wfm/team.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass, field
+
+from libzapi.domain.shared_objects.logical_key import LogicalKey
+
+
+@dataclass(frozen=True, slots=True)
+class Team:
+    id: str
+    name: str
+    description: str = ""
+    manager_id: int | None = None
+    agents_ids: list[str] = field(default_factory=list)
+    is_deleted: bool = False
+    deleted_at: str | None = None
+    tymeshift_account_id: int | None = None
+
+    @property
+    def logical_key(self) -> LogicalKey:
+        base = self.name.lower().replace(" ", "_")
+        return LogicalKey("wfm_team", base)
+
+
+@dataclass(frozen=True, slots=True)
+class BulkAgentsResult:
+    status: str = ""
+    affected_teams: list[str] = field(default_factory=list)