Quality of Life on client.extraction_pipelines.runs.list

CHANGELOG.md

@@ -17,6 +17,12 @@ Changes are grouped as follows
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+## [6.28.5] - 2023-10-03
+### Improved
+- Quality of life improvement to `client.extraction_pipelines.runs.list` method. It uses `Literal` in the type hint
+  of the `statuses` parameter, allows a single `statuses` parameter, and accepts `created_time` on the format `12-ago`
+  (same as the `DatapointAPI` methods).
+
 ## [6.28.4] - 2023-10-03
 ### Fixed
 - Overload data_set/create for improved type safety

cognite/client/_api/extractionpipelines.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Literal, Sequence, overload
 
+from typing_extensions import TypeAlias
+
 from cognite.client import utils
 from cognite.client._api_client import APIClient
 from cognite.client._constants import DEFAULT_LIMIT_READ
@@ -17,13 +20,18 @@
     TimestampRange,
 )
 from cognite.client.data_classes.extractionpipelines import StringFilter
+from cognite.client.utils import datetime_to_ms
 from cognite.client.utils._identifier import IdentifierSequence
+from cognite.client.utils._time import time_ago_to_ms
 
 if TYPE_CHECKING:
     from cognite.client import CogniteClient
     from cognite.client.config import ClientConfig
 
 
+RunStatus: TypeAlias = Literal["success", "failure", "seen"]
+
+
 class ExtractionPipelinesAPI(APIClient):
     _RESOURCE_PATH = "/extpipes"
 
@@ -217,23 +225,28 @@ class ExtractionPipelineRunsAPI(APIClient):
     def list(
         self,
         external_id: str,
-        statuses: Sequence[str] | None = None,
+        statuses: RunStatus | Sequence[RunStatus] | Sequence[str] | None = None,
         message_substring: str | None = None,
-        created_time: dict[str, Any] | TimestampRange | None = None,
+        created_time: dict[str, Any] | TimestampRange | str | None = None,
         limit: int | None = DEFAULT_LIMIT_READ,
     ) -> ExtractionPipelineRunList:
         """`List runs for an extraction pipeline with given external_id <https://developer.cognite.com/api#tag/Extraction-Pipelines/operation/filterRuns>`_
 
         Args:
             external_id (str): Extraction pipeline external Id.
-            statuses (Sequence[str] | None): One or more among "success" / "failure" / "seen".
+            statuses (RunStatus | Sequence[RunStatus] | Sequence[str] | None): One or more among "success" / "failure" / "seen".
             message_substring (str | None): Failure message part.
-            created_time (dict[str, Any] | TimestampRange | None): The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
+            created_time (dict[str, Any] | TimestampRange | str | None): The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
             limit (int | None): Maximum number of ExtractionPipelines to return. Defaults to 25. Set to -1, float("inf") or None to return all items.
 
         Returns:
             ExtractionPipelineRunList: List of requested extraction pipeline runs
 
+        Tip:
+            The created_time paremeter supports in addition to a dictonary with the format
+            `{"min": epoch_min, "max": epoch_max}`, arguments given on the format `<integer>(s|m|h|d|w)-ago`.
+            For example, `12h-ago`, which will be parsed to `{"min"= now - 12h-ago}`.
+
         Examples:
 
             List extraction pipeline runs::
@@ -247,12 +260,23 @@ def list(
                 >>> from cognite.client import CogniteClient
                 >>> c = CogniteClient()
                 >>> runsList = c.extraction_pipelines.runs.list(external_id="test ext id", statuses=["seen"], statuslimit=5)
+
+            Get all failed pipeline runs the last 24 hours for pipeliene 'extId':
+
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes import ExtractionPipelineRun
+                >>> c = CogniteClient()
+                >>> res = c.extraction_pipelines.runs.list(external_id="extId", statuses="failure", created_time="24h-ago")
         """
+        if isinstance(created_time, str):
+            timespan = time_ago_to_ms(created_time)
+            now = datetime_to_ms(datetime.now(timezone.utc))
+            created_time = TimestampRange(min=now - timespan)
 
         if statuses is not None or message_substring is not None or created_time is not None:
             filter = ExtractionPipelineRunFilter(
                 external_id=external_id,
-                statuses=statuses,
+                statuses=[statuses] if isinstance(statuses, str) else statuses,
                 message=StringFilter(substring=message_substring),
                 created_time=created_time,
             ).dump(camel_case=True)

cognite/client/_version.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
-__version__ = "6.28.4"
+__version__ = "6.28.5"
 __api_subversion__ = "V20220125"

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "cognite-sdk"
 
-version = "6.28.4"
+version = "6.28.5"
 description = "Cognite Python SDK"
 readme = "README.md"
 documentation = "https://cognite-sdk-python.readthedocs-hosted.com"

tests/tests_integration/test_api/test_extraction_pipelines.py

@@ -1,14 +1,17 @@
+from datetime import datetime, timedelta, timezone
+
 import pytest
 
 from cognite.client import CogniteClient
 from cognite.client.data_classes import ExtractionPipeline, ExtractionPipelineRun, ExtractionPipelineUpdate
-from cognite.client.data_classes.extractionpipelines import ExtractionPipelineContact
+from cognite.client.data_classes.extractionpipelines import ExtractionPipelineContact, ExtractionPipelineRunList
 from cognite.client.exceptions import CogniteNotFoundError
+from cognite.client.utils import datetime_to_ms
 from cognite.client.utils._text import random_string
 
 
 @pytest.fixture
-def new_extpipe(cognite_client):
+def new_extpipe(cognite_client: CogniteClient):
     testid = random_string(50)
     dataset = cognite_client.data_sets.list()[0]
     extpipe = cognite_client.extraction_pipelines.create(
@@ -33,6 +36,28 @@ def new_extpipe(cognite_client):
     assert cognite_client.extraction_pipelines.retrieve(extpipe.id) is None
 
 
+@pytest.fixture
+def populated_runs(cognite_client: CogniteClient, new_extpipe: ExtractionPipeline) -> ExtractionPipelineRunList:
+    now = datetime_to_ms(datetime.now(timezone.utc))
+    a_year_ago = datetime_to_ms(datetime.now(timezone.utc).replace(year=datetime.now(timezone.utc).year - 1))
+    runs = [
+        ExtractionPipelineRun(
+            extpipe_external_id=new_extpipe.external_id, status="failure", message="lorem ipsum", created_time=now
+        ),
+        ExtractionPipelineRun(
+            extpipe_external_id=new_extpipe.external_id,
+            status="success",
+            message="dolor sit amet",
+            created_time=a_year_ago,
+        ),
+    ]
+    created = []
+    for run in runs:
+        new_run = cognite_client.extraction_pipelines.runs.create(run)
+        created.append(new_run)
+    return ExtractionPipelineRunList(created)
+
+
 class TestExtractionPipelinesAPI:
     def test_retrieve(self, cognite_client):
         res = cognite_client.extraction_pipelines.list(limit=1)
@@ -116,3 +141,32 @@ def test_list_extraction_pipeline_runs(
         dumped = res.dump()
         for run in dumped:
             assert run["external_id"] == new_extpipe.external_id
+
+    def test_list_failed_extraction_pipeline_runs(
+        self,
+        cognite_client: CogniteClient,
+        new_extpipe: ExtractionPipeline,
+        populated_runs: ExtractionPipelineRunList,
+    ) -> None:
+        expected = ExtractionPipelineRunList([run for run in populated_runs if run.status == "failure"])
+
+        filtered = cognite_client.extraction_pipelines.runs.list(
+            external_id=new_extpipe.external_id, statuses="failure", limit=1
+        )
+
+        assert expected.dump() == filtered.dump()
+
+    def test_filter_extraction_pipeline_runs_created_ago(
+        self,
+        cognite_client: CogniteClient,
+        new_extpipe: ExtractionPipeline,
+        populated_runs: ExtractionPipelineRunList,
+    ) -> None:
+        yesterday = datetime_to_ms(datetime.now(timezone.utc) - timedelta(days=1))
+        expected = ExtractionPipelineRunList([run for run in populated_runs if run.created_time > yesterday])
+
+        filtered = cognite_client.extraction_pipelines.runs.list(
+            external_id=new_extpipe.external_id, created_time="24h-ago", limit=1
+        )
+
+        assert expected.dump() == filtered.dump()