Metadata-Version: 2.5
Name: improvado-pipeline-sdk
Version: 0.16.0
Summary: SDK for authoring Improvado custom pipeline workflows on Temporal
License-Expression: Apache-2.0
Keywords: etl,pipeline,temporal,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: aiohttp>=3.9
Requires-Dist: boto3>=1.34
Requires-Dist: clickhouse-connect>=0.7
Requires-Dist: nexus-rpc>=1.1
Requires-Dist: pyright>=1.1
Requires-Dist: ruff>=0.4
Requires-Dist: structlog>=23.1
Requires-Dist: temporalio>=1.26
Requires-Dist: zstandard>=0.22
Description-Content-Type: text/markdown

# improvado-pipeline-sdk

SDK for authoring custom data pipeline workflows on [Temporal](https://temporal.io).

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

## Install

```bash
pip install improvado-pipeline-sdk
```

## Usage

### Workflow code

```python
import dataclasses
from temporalio import workflow

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


@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

```python
from temporalio import activity
from pipeline_sdk.runtime import (
    load_credentials,
    get_current_tenant,
    PipelineState,
)


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

### Run events

Declare the source namespace and record events inside activities. The interceptor
flushes at activity exit, including failures:

```python
from pipeline_sdk.events import objects

campaign = objects.define(
    'campaign', system='meta_ads', scope='act_123', label='Campaign',
    link='https://adsmanager.facebook.com/?ids={key}',
)

@activity.defn(name='sync_campaigns')
async def sync_campaigns(params) -> dict:
    campaign.created('456', 'Campaign created', name='Q4 Paid Search',
                     metrics={'budget': 100.5}, objective='CONVERSIONS')
    return {'created': 1}
```

Version 1 records `schema_version`, `dedup_key`, `occurred_at`, `kind`, `title`,
`object {system, scope?, type, key, name?}`, `metrics` and `data`. Kinds are
`created`, `updated`, `deleted`, `sent`, `skipped`, `warning`; the generic
`campaign.event(kind, key, title, ...)` accepts the same fields. Emit `skipped`
only when the user explicitly requests skipped-operation events. Do not record
routine no-ops, unchanged objects, already-synced records or dry-run intentions by
default; an empty event list is valid when nothing happened.

Object identity is workspace + system + scope + type + key, independent of the
pipeline and connection used to access it. Use the source account/organization
as scope; omit scope only for globally unique IDs within that system and type.
`label` names the type, while `name` names this particular object. Metrics are
arbitrary finite numbers for display; the platform does not automatically sum
them. Other keyword arguments become `data`, bounded to 8 KiB of UTF-8 JSON.

Pass the `PipelineCredentials` returned by `prepare_credentials` as a direct
activity argument when that activity emits events. The event interceptor resolves
its DTS session automatically, without requiring the business code to call
`load_credentials` just for reporting. An activity that already calls
`load_credentials` can use that bound session instead. Credentials are never
propagated through Temporal headers or automatically added to other activities
or child workflows; pass required credentials explicitly through their arguments.
This uses the existing activity-argument/data-converter path, not a claim that
all Temporal payloads are encrypted.

A user session is currently required; workspace-only credentials cannot authenticate
this endpoint. Delivery retries 429, 5xx and network failures. Exhausted delivery
is logged and counted without failing the business activity. Every activity's
`pipeline_sdk.events.delivery` log carries `run_id`, `activity_id`, sent/dropped
counts and reasons. Author return values are preserved, including dicts; delivery
diagnostics are not injected into them. Workers send a separate per-activity/per-attempt
report using the reserved Temporal signal `__pipeline_sdk_event_delivery_v1` only
when something was sent or dropped, including when the event POST fails. Empty
activities send no signal and add no reporting RPC delay or history entries. Workflow Manager exposes
`event_delivery` alongside `result`: status `reported` or `unknown`, `reports`
with `activity_id`, `attempt`, `sent`, `dropped`, `reasons`, and
`unknown_activity_ids`. Duplicate reports are deduplicated per attempt; retry
attempts are not summed. `reported` means Workflow Manager parsed the complete
history, even when some or all activities have no report. `unknown_activity_ids`
is informational: missing reports may mean no events, an older worker, or failed
publication; failure is logged as `pipeline_sdk.events.report_failed`. This status
does not certify delivery or worker reporting support. Running runs, truncated
history, or history-parse failure remain `unknown`. The signal uses the configured Temporal
client/data converter, is pinned to the workflow execution, contains no credentials,
and has a five-second RPC timeout. Failure to publish never changes the business
result. Custom workers should register `RunEventsInterceptor(client)` and reserve
this signal name. Recording outside an activity raises an
explicit error.

Adhoc runs omit pipeline_id and keep their events. Automatic deduplication combines
execution ID, activity ID, event content and occurrence, so retry reordering does
not duplicate unchanged events and new executions do not collide. For a logical
event whose content changes across retries, supply your own stable `dedup_key`;
include execution identity when different executions should remain distinct.

`python -m pipeline_sdk validate code.py` returns `declared_objects` from the SDK
registry after successful smoke validation; the platform stores this metadata
instead of parsing Python. With `--no-smoke` or a failed validation, object
declarations are not inspected: JSON reports `declared_objects_inspected: false`
and omits `declared_objects`. An empty registry after successful import produces
a C9 warning, never a validation error; event calls are not statically inspected.
The platform retains all events from the last 30 days plus the ten newest live
events per source object (agency/workspace + system + scope + type + key), across
pipelines. Corrections expire with their event; older run event lists may be
incomplete. Deleted/superseded rows do not count toward this minimum and can be
purged once both occurrence and deletion are older than 30 days. Delivery reports
describe original delivery, independently of retention. Deleting in the UI is a
soft delete that is reversible until retention cleanup. Deploy this SDK release
before the platform contract update.

### Worker / client setup

```python
from pipeline_sdk.tenant import (
    TenantInterceptor,       # worker side
    TenantClientInterceptor, # client side
    TenantID,
)
```

### Run a workflow locally against production Temporal

For inner-loop debugging — write your workflow on your laptop and run
it against production Temporal under your tenant identity. Activities
defined in your file run on the laptop; cross-queue activities such as
`les_extract`, `ch_load`, `prepare_credentials` are dispatched to
Improvado shared queues and execute on Improvado infra.

```bash
pip install improvado-pipeline-sdk

# Mint a Temporal JWT — ask your platform team for the runbook.
export TEMPORAL_API_URL=flow-api.tools.improvado.io:443
export TEMPORAL_AUTH_JWT=$(cat /tmp/temporal-token)
export TENANT_AGENCY_UUID=fc010783-7a09-449b-84ab-0c1660b15245
export TENANT_WORKSPACE_ID=42
# optional: TENANT_USER_ID=99       # ADR-010 user-scope tenants
# optional: TEMPORAL_TLS=false      # debug only — local docker-compose
# optional: VAULT_ADDR / VAULT_MOUNT_POINT — point load_credentials at prod
```

**One-shot mode (default)** — point at a workflow file, the runner
spins up a worker on the tenant queue, starts the workflow, prints
the result, and exits:

```bash
python -m pipeline_sdk hello_sdk.py
python -m pipeline_sdk hello_sdk.py --params '{"label": "hi"}'
python -m pipeline_sdk ./pipelines/ --name HelloSdkWorkflow --params-file params.json
```

**Worker-only mode** — long-running worker that polls the tenant
queue `tenant-{agency}-{workspace}[-{user}]`. Trigger runs through
Workflow Manager API or Temporal CLI:

```bash
python -m pipeline_sdk ./pipelines/ --worker-only
# or use the dedicated entry: improvado-pipeline-run ./pipelines/ --worker-only
```

`Ctrl-C` drains and exits in both modes.

### Type checking

```bash
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.types` | Pure, Temporal-serializable data types and enums — `DataRef`, `TenantID`, `Cluster`, `DateRangeRequest`, `DateRangeType`, `PipelineCredentials`, `PipelineSecret`, `S3Credentials`, `StorageCredentials`, `LesActivityWithS3Result`, `PipelineLoadResult`. Safe to import from either workflow or activity code. |
| `pipeline_sdk.activities` | Predefined workflow-side wrappers — `les_extract`, `ch_load`, `prepare_credentials`, `cleanup_credentials`. Import inside `workflow.unsafe.imports_passed_through()`. Custom activities defined in your own module inherit the workflow's task queue — do not pass `task_queue` to `workflow.execute_activity`. |
| `pipeline_sdk.runtime` | Activity-side helpers — `load_credentials`, `PipelineState`, `get_current_tenant`, plus `VAULT_ADDR` / `VAULT_MOUNT_POINT` for Vault config overrides. |
| `pipeline_sdk.events` | Run events — `objects.define`, the six `ObjectType` methods, `flush_events`, `RunEventsInterceptor`. `define` is pure (workflow-safe); recording and flushing are activity-side. |
| `pipeline_sdk.tenant` | Tenant propagation infrastructure — `TenantInterceptor`, `TenantClientInterceptor`, `build_tenant_headers`, plus `TenantID` / `get_current_tenant` re-exports. |
