improvado-pipeline-sdk 0.1.0

SDK for authoring Improvado custom pipeline workflows on Temporal

improvado-pipeline-sdk

SDK for authoring custom data pipeline workflows on Temporal.

Provides type definitions, helper functions, and constants for building ETL/ELT pipelines that run on Improvado's Temporal infrastructure.

Install

pip install improvado-pipeline-sdk

Usage

Workflow code

import dataclasses
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from pipeline_sdk.workflow import (
        les_extract,
        DataRef,
        DateRangeRequest,
        DateRangeType,
        prepare_credentials,
        cleanup_credentials,
        CUSTOM_ACTIVITY_QUEUE,
    )


@dataclasses.dataclass(frozen=True)
class MyPipelineParams:
    connector_id: int


@workflow.defn(sandboxed=False, name="MyPipeline")
class MyPipelineWorkflow:
    @workflow.run
    async def run(self, params: MyPipelineParams) -> dict:
        result = await les_extract(
            connector_id=params.connector_id,
            data_source="facebook",
            report_type="ad_insights",
            date_range=DateRangeRequest(
                date_range_type=DateRangeType.MANUAL,
                params={"date_from": "2026-01-01", "date_to": "2026-03-31"},
            ),
        )
        return {"data_ref": result.data_ref.uri}

Activity code

from temporalio import activity
from pipeline_sdk.activity import read_pipeline_secret, get_current_tenant


@activity.defn
async def my_activity(creds) -> dict:
    tenant = get_current_tenant()
    secret = await read_pipeline_secret(creds)
    # secret.s3, secret.storage, secret.connections
    return {"tenant": tenant.agency_uuid}

Type checking

pyright my_workflow.py
ruff check my_workflow.py
ruff format my_workflow.py

Both pyright and ruff are included as dependencies.

Package structure

Module	Purpose
pipeline_sdk.workflow	Types and helpers for @workflow.defn code (sandbox-safe)
pipeline_sdk.activity	Types and helpers for @activity.defn code (I/O)
pipeline_sdk.schedule	@scheduled decorator for cron workflows
pipeline_sdk.types	All shared dataclasses and enums
pipeline_sdk.credentials	Credential types and Vault helpers