SDK

frank-sdk is the Python contract for custom transform authors. It gives Python-runner patterns a stable way to read runtime config, query data, emit metrics and lineage, and return structured results to the platform.

Install

From the shared utilities package:

cd utils

# Core SDK
pip install -e ".[sdk]"

# With Trino support
pip install -e ".[sdk-trino]"

# With pandas and pyarrow
pip install -e ".[sdk-pandas]"

# Full local data stack
pip install -e ".[sdk-full]"

In packaged environments:

pip install "frank-shared[sdk-full]"

Runtime contract

Python-runner transforms receive configuration through TRANSFORM_CONFIG.

{
  "artifact_id": "uuid",
  "run_id": "uuid",
  "tenant_id": "uuid",
  "source_table": "iceberg.bronze.orders",
  "source_tables": ["iceberg.bronze.orders"],
  "target_table": "iceberg.silver.orders_clean",
  "output_contract": {
    "creates_table": true,
    "output_type": "table"
  },
  "params": {
    "filter_expression": "status != 'deleted'"
  },
  "cursors": {
    "input_cursors": {
      "iceberg.bronze.orders": {
        "mode": "delta",
        "cursor_field": "_extracted_at",
        "cursor_value": "2026-02-01T00:00:00Z"
      }
    },
    "cutoff_cursors": {
      "iceberg.bronze.orders": {
        "cursor_value": "2026-02-04T12:00:00Z"
      }
    }
  }
}

source_table, source_tables, and target_table are control-plane-owned identities. A table-producing pattern must write only ctx.target_table; a similarly named value in params is not an output override. The backend verifies the declared Iceberg table and snapshot before it records the run as completed.

A pattern that intentionally produces no table declares output_contract.creates_table: false in its pattern metadata. Its ctx.target_table is None; it completes through logs and FrankResult without an Iceberg snapshot or external Dagster data asset.

The transform writes one final FrankResult JSON object to stdout. Logs, metrics, progress, and lineage should go through SDK emitters.

Minimal transform

from frank_sdk import FrankContext, FrankResult

def main():
    ctx = FrankContext.from_env()
    result = FrankResult.success(
        output_row_count=0,
        metrics={
            "mode": ctx.get_param("mode", "default"),
            "target_table": ctx.target_table,
        },
    )
    result.write_to_stdout()

if __name__ == "__main__":
    main()

Querying with Trino

from frank_sdk import FrankContext, FrankResult, emit_metric, emit_lineage

def main():
    ctx = FrankContext.from_env()
    conn = ctx.get_trino_connection()
    cur = conn.cursor()

    where = ctx.build_incremental_filter(ctx.source_table)
    cur.execute(f"""
        CREATE TABLE {ctx.target_table} AS
        SELECT *
        FROM {ctx.source_table}
        WHERE {where}
    """)

    cur.execute(f"SELECT COUNT(*) FROM {ctx.target_table}")
    count = cur.fetchone()[0]

    emit_metric("rows_processed", count)
    emit_lineage(
        source=ctx.source_table,
        target=ctx.target_table,
        operation="copy",
    )

    FrankResult.success(output_row_count=count).write_to_stdout()

if __name__ == "__main__":
    main()

Connection environment variables:

Variable	Default
TRINO_HOST	localhost
TRINO_PORT	8080
TRINO_USER	frank-transform
TRINO_CATALOG	iceberg
TRINO_SCHEMA	unset
TRINO_PASSWORD	unset

FrankContext

Important properties:

Property	Meaning
artifact_id	Transform artifact UUID.
run_id	Transform run UUID.
tenant_id	Tenant UUID.
source_table	Primary input table.
source_tables	All input tables.
target_table	Canonical control-plane-owned output table, or None for an explicit log-only pattern.
raw_config["output_contract"]	Declares whether the run creates a physical table.
params	Pattern or transform parameters.
cursors	Input and cutoff cursor state.
raw_config	Full runtime config.

Useful methods:

ctx.get_param("key", default=None)
ctx.require_param("key")
ctx.get_trino_connection()
ctx.get_input_cursor(table)
ctx.get_cutoff_cursor(table)
ctx.build_incremental_filter(table)

FrankResult

Use factory helpers:

FrankResult.success(output_row_count=100, input_row_count=120)
FrankResult.failure("Invalid source data", {"column": "id"})
FrankResult.partial(output_row_count=80, error_message="20 rows skipped")

Add metrics and data quality:

from frank_sdk import FrankResult, DataQualityResult

result = FrankResult.success(output_row_count=100)
result.add_metric("duplicates_removed", 5)
result.add_data_quality_result(
    DataQualityResult(
        rule_name="non_null_id",
        passed=True,
        total_rows=100,
        failed_rows=0,
    )
)
result.write_to_stdout()

Output fields:

Field	Meaning
status	success, failure, or partial.
output_row_count	Rows written.
input_row_count	Rows read.
rows_affected	Rows changed by update/delete style operations.
metrics	Custom JSON-serializable metrics.
data_quality	Data quality check results.
error_message	Human-readable error.
error_details	Structured error context.
output_snapshot_id	Iceberg snapshot ID serialized as an exact decimal string; integer constructor input remains accepted.

Structured logging, metrics, and lineage

from frank_sdk import emit_log, emit_warning, emit_metric, emit_progress, emit_lineage

emit_log("Starting transform", context={"table": "orders"})
emit_metric("rows_processed", 1000, tags={"stage": "clean"})
emit_progress(current=5, total=10, message="Processed batch 5")
emit_warning("Skipped rows with invalid dates", {"count": 3})
emit_lineage(
    source=["iceberg.bronze.orders", "iceberg.bronze.customers"],
    target="iceberg.gold.customer_orders",
    operation="join",
    columns={"customer_id": ["orders.customer_id", "customers.id"]},
)

Emitters write structured records to stderr for platform capture while keeping stdout reserved for the final FrankResult.

Local testing

Pair the SDK with the Python pattern CLI:

frank init my-pattern --template python
cd patterns/my-pattern
frank validate .
frank test . --config tests/sample_config.json --env-file .env.test

Use test fixtures in tests/sample_config.json to create Trino tables, run assertions, and clean up after the transform.