List runs for an extraction pipeline

async AsyncCogniteClient.extraction_pipelines.runs.list(
external_id: str,
statuses: Literal['success', 'failure', 'seen'] | Sequence[Literal['success', 'failure', 'seen']] | SequenceNotStr[str] | None = None,
message_substring: str | None = None,
created_time: dict[str, Any] | TimestampRange | str | None = None,
limit: int | None = 25,
) → ExtractionPipelineRunList

List runs for an extraction pipeline with given external_id.

Parameters:

  • external_id (str) – Extraction pipeline external Id.

  • statuses (RunStatus | Sequence[RunStatus] | SequenceNotStr[str] | None) – One or more among “success” / “failure” / “seen”.

  • message_substring (str | None) – Failure message part.

  • created_time (dict[str, Any] | TimestampRange | str | None) – Range between two timestamps. Possible keys are min and max, with values given as timestamps in ms. If a string is passed, it is assumed to be the minimum value.

  • limit (int | None) – Maximum number of ExtractionPipelines to return. Defaults to 25. Set to -1, float(“inf”) or None to return all items.

Returns:

List of requested extraction pipeline runs

Return type:

ExtractionPipelineRunList

Tip

The created_time parameter can also be passed as a string, to support the most typical usage pattern of fetching the most recent runs, meaning it is implicitly assumed to be the minimum created time. The format is “N[timeunit]-ago”, where timeunit is w,d,h,m (week, day, hour, minute), e.g. “12d-ago”.

Examples

List extraction pipeline runs:

>>> from cognite.client import CogniteClient, AsyncCogniteClient
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> runsList = client.extraction_pipelines.runs.list(external_id="test ext id", limit=5)

Filter extraction pipeline runs on a given status:

>>> runs_list = client.extraction_pipelines.runs.list(
...     external_id="test ext id", statuses=["seen"], limit=5
... )

Get all failed pipeline runs in the last 24 hours for pipeline ‘extId’:

>>> from cognite.client.data_classes import ExtractionPipelineRun
>>> res = client.extraction_pipelines.runs.list(
...     external_id="extId", statuses="failure", created_time="24h-ago"
... )