Manually Poll a Source | Validio SDK Recipes

Overview

A source doesn’t have to run on a schedule, and can instead be polled manually. This is useful in a pipeline or orchestration tool, such as Airflow, that first loads data that should then be analyzed. The poll_source helper triggers a manual poll, waits for it to complete, and reports whether the poll produced any new incidents, all in a single call. It returns a SourcePollResult that you can branch on to build a circuit breaker that stops the pipeline when fresh data fails its validators. For more on circuit breakers, see Circuit breakers.

Steps

Set up the client

Import poll_source from validio_sdk and create a client. The client picks up your credentials automatically from environment variables or a config.json, the same way as the rest of the SDK. See How the CLI and SDK find credentials. The poll_source helper requires Validio SDK version 10.0.3 or later.

Poll and check for incidents

Call poll_source with the source’s ID or resource name. The helper triggers the poll, waits for it to complete, and then checks the incidents API for any incidents created since the poll started. Set timeout_seconds to cap how long to wait, and pass a namespace if you identify the source by resource name instead of ID.

Break on a failed poll

Once the poll has started, errors are returned on result.error rather than raised, so a failed poll doesn’t crash your task unexpectedly. Check it first, and treat a poll that couldn’t complete as a broken circuit.

Break on new incidents

result.has_incidents is True when the poll produced new incidents. Raise an error (or return a non-zero exit code) so your orchestrator halts the pipeline; otherwise, let it continue. Adjust this branch to apply your own threshold, for example to allow a small number of low-severity incidents.

Full code sample

Python

import os

from validio_sdk import poll_source
from validio_sdk._api.api import APIClient


async def example(source_id: str = os.environ["EXAMPLE_SOURCE_ID"]):
    client = APIClient()

    # Trigger a manual poll, wait for it to finish, and check whether it
    # produced any new incidents — all in a single call.
    result = await poll_source(
        source_id_or_name=source_id,
        client=client,
        timeout_seconds=300,
    )

    # The poll never started, failed while running, or timed out. Treat this as
    # a broken circuit so bad data can't slip through unchecked.
    if result.error is not None:
        raise Exception(f"Source poll did not complete: {result.error}")

    # The poll completed. Break the pipeline if it produced new incidents.
    if result.has_incidents:
        raise Exception("New incidents detected — stopping the pipeline")

    print("No new incidents — the pipeline can continue")

Overview

Getting Started

Validio API

Validio CLI

Validio SDK

Validio Code (IaC)

Manually Poll a Source and Its Validators

Overview

Steps

Full code sample

​Overview

​Steps

​Full code sample

Overview

Steps

Full code sample