Working hours

Author: KirillSergeevich

linkedapi/__init__.py

@@ -78,7 +78,7 @@
 from linkedapi.types import __all__ as _types_all
 from linkedapi.webhooks import parse_webhook_event
 
-__version__ = "1.3.5"
+__version__ = "1.3.6"
 PredefinedOperation = Operation
 
 __all__ = [

linkedapi/core/operation.py

@@ -60,6 +60,7 @@ def status(self, workflow_id: str) -> WorkflowInProgressResponse | MappedRespons
                 workflow_id=workflow_id,
                 workflow_status=workflow_result.workflow_status,
                 message=workflow_result.message,
+                pending_reason=workflow_result.pending_reason,
             )
 
         completion = self._get_completion(workflow_result)

linkedapi/errors.py

@@ -66,6 +66,8 @@
     "linkedinAccountSignedOut",
     "languageNotSupported",
     "workflowTimeout",
+    "outsideWorkingHours",
+    "workingHoursWaitExpired",
     "httpError",
     "tooManyRequests",
     "accountNotFound",
@@ -87,6 +89,8 @@
     "linkedinAccountSignedOut",
     "languageNotSupported",
     "workflowTimeout",
+    "outsideWorkingHours",
+    "workingHoursWaitExpired",
     "httpError",
     "tooManyRequests",
     "accountNotFound",

linkedapi/types/__init__.py

@@ -235,6 +235,7 @@
     WorkflowFailure,
     WorkflowInProgressResponse,
     WorkflowInProgressStatus,
+    WorkflowPendingReason,
     WorkflowResponse,
     WorkflowStartedResponse,
     WorkflowStatus,
@@ -448,6 +449,7 @@
     "WorkflowFailure",
     "WorkflowInProgressResponse",
     "WorkflowInProgressStatus",
+    "WorkflowPendingReason",
     "WorkflowResponse",
     "WorkflowStartedResponse",
     "WorkflowStatus",

linkedapi/types/workflow.py

@@ -9,6 +9,7 @@
 WorkflowRunningStatus = Literal["running"]
 WorkflowInProgressStatus = Literal["pending", "running"]
 WorkflowStatus = Literal["pending", "running", "completed", "failed"]
+WorkflowPendingReason = Literal["queued", "outsideWorkingHours"]
 WorkflowDefinition = dict[str, Any] | list[dict[str, Any]]
 WorkflowCompletion = dict[str, Any] | list[dict[str, Any]]
 
@@ -22,12 +23,15 @@ class WorkflowStartedResponse(LinkedApiModel):
     workflow_id: str
     workflow_status: WorkflowInProgressStatus
     message: str | None = None
+    # Optional so an older API that does not send it still parses.
+    pending_reason: WorkflowPendingReason | None = None
 
 
 class WorkflowInProgressResponse(LinkedApiModel):
     workflow_id: str
     workflow_status: WorkflowInProgressStatus
     message: str | None = None
+    pending_reason: WorkflowPendingReason | None = None
 
 
 class WorkflowFailure(LinkedApiModel):
@@ -43,6 +47,7 @@ class WorkflowResponse(LinkedApiModel):
     workflow_id: str
     workflow_status: WorkflowStatus
     message: str | None = None
+    pending_reason: WorkflowPendingReason | None = None
     completion: WorkflowCompletion | None = None
     failure: WorkflowFailure | None = None
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "linkedapi"
-version = "1.3.5"
+version = "1.3.6"
 description = "Official synchronous Python SDK for Linked API."
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_errors.py

@@ -21,6 +21,8 @@ def test_error_type_sets_match_node_contract() -> None:
         "linkedinAccountSignedOut",
         "languageNotSupported",
         "workflowTimeout",
+        "outsideWorkingHours",
+        "workingHoursWaitExpired",
         "httpError",
         "tooManyRequests",
         "accountNotFound",

tests/test_working_hours_contract.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import Mock
+
+import pytest
+
+from linkedapi.core.operation import Operation
+from linkedapi.errors import LINKED_API_ERROR_TYPES
+from linkedapi.types import (
+    WorkflowInProgressResponse,
+    WorkflowResponse,
+    WorkflowStartedResponse,
+)
+
+
+class _StubMapper:
+    def map_request(self, params: Any) -> Any:
+        return params
+
+    def map_response(self, completion: Any) -> Any:
+        return completion
+
+
+class _StubOperation(Operation):  # type: ignore[type-arg]
+    operation_name = "fetchPerson"
+
+    def __init__(self, http_client: Any) -> None:
+        super().__init__(http_client)
+        self.mapper = _StubMapper()  # type: ignore[assignment]
+
+
+def _operation(workflow_result: dict[str, Any]) -> _StubOperation:
+    http_client = Mock()
+    http_client.get.return_value = Mock(
+        error=None,
+        result=WorkflowResponse.model_validate(workflow_result),
+    )
+    return _StubOperation(http_client)
+
+
+class TestPendingReasonModels:
+    def test_in_progress_response_accepts_a_pending_reason(self) -> None:
+        response = WorkflowInProgressResponse(
+            workflowId="wf-1",
+            workflowStatus="pending",
+            pendingReason="outsideWorkingHours",
+        )
+
+        assert response.pending_reason == "outsideWorkingHours"
+
+    def test_started_response_accepts_a_pending_reason(self) -> None:
+        response = WorkflowStartedResponse(
+            workflowId="wf-1",
+            workflowStatus="pending",
+            pendingReason="queued",
+        )
+
+        assert response.pending_reason == "queued"
+
+    @pytest.mark.parametrize(
+        "model",
+        [WorkflowInProgressResponse, WorkflowStartedResponse, WorkflowResponse],
+    )
+    def test_field_is_optional_for_back_compat(self, model: Any) -> None:
+        response = model(workflowId="wf-1", workflowStatus="running")
+
+        assert response.pending_reason is None
+
+
+class TestStatusCarriesPendingReason:
+    def test_carries_the_outside_working_hours_reason(self) -> None:
+        operation = _operation(
+            {
+                "workflowId": "wf-1",
+                "workflowStatus": "pending",
+                "pendingReason": "outsideWorkingHours",
+            }
+        )
+
+        result = operation.status("wf-1")
+
+        assert isinstance(result, WorkflowInProgressResponse)
+        assert result.pending_reason == "outsideWorkingHours"
+
+    def test_carries_the_queued_reason(self) -> None:
+        operation = _operation(
+            {"workflowId": "wf-1", "workflowStatus": "pending", "pendingReason": "queued"}
+        )
+
+        result = operation.status("wf-1")
+
+        assert isinstance(result, WorkflowInProgressResponse)
+        assert result.pending_reason == "queued"
+
+    def test_keeps_none_when_the_api_sends_nothing(self) -> None:
+        operation = _operation({"workflowId": "wf-1", "workflowStatus": "running"})
+
+        result = operation.status("wf-1")
+
+        assert isinstance(result, WorkflowInProgressResponse)
+        assert result.pending_reason is None
+
+
+class TestWorkingHoursErrorTypes:
+    @pytest.mark.parametrize("error_type", ["outsideWorkingHours", "workingHoursWaitExpired"])
+    def test_error_type_is_known(self, error_type: str) -> None:
+        assert error_type in LINKED_API_ERROR_TYPES