Paul Hudson

Appearance

Book a call

Connect ​

connect writes skippr.yml. Python Config.save(), skipprd connect, and a hand-written file are the same persist path. There is no second field catalog. Secrets persist as ${ENV} references, never plaintext.

--pipeline and --name are required, and connect wires that entry into the pipeline. Existing skippr.yml or skippr.yaml is reused; a new file starts empty. Each field you pass replaces that field (a flattened nested flag such as --object-store-type replaces the whole object_store block), and sibling pipelines and other entries are kept. Python registration and Config.save() use the same merge.

Root skippr: keys are global flags, not a connect skippr verb.

Workspace ​

python
from skippr import Config, DataSourceS3, LocalStorage, Pipeline

cfg = Config().workspace("quickstart").storage(LocalStorage())
src = cfg.data_source("sample", DataSourceS3(s3_bucket="skippr-public-sample-data", s3_prefix="bike-hire"))
cfg.pipeline("bikehire", Pipeline(data_source=src))
cfg.save("skippr.yml")
bash
skipprd --workspace quickstart --storage-mode local connect data-source s3 \
  --pipeline bikehire \
  --name sample \
  --s3-bucket skippr-public-sample-data \
  --s3-prefix bike-hire
yaml
skippr:
  workspace: quickstart
  skipprd_el_storage_mode: local

pipelines:
  bikehire:
    data_source: data_sources.sample

data_sources:
  sample:
    S3:
      s3_bucket: skippr-public-sample-data
      s3_prefix: bike-hire

--storage-mode is local or s3. --store-type is sled, dynamodb, or cloud-tables; --store-name is the SkipprStore table. WAL backend stays --wal-storage / WAL_STORAGE (not a YAML key).

Python classes take required fields as required keyword arguments. Config.save() merge-writes like the CLI. Secret fields take EnvRef.

Data source ​

python
from skippr import Config, DataSourceS3, Pipeline

cfg = Config.discover()
src = cfg.data_source(
    "sample",
    DataSourceS3(
        s3_bucket="skippr-public-sample-data",
        s3_prefix="bike-hire",
        region="us-east-1",
        format="json",
    ),
)
cfg.pipeline("bikehire", Pipeline(data_source=src))
cfg.save()
bash
skipprd connect data-source s3 \
  --pipeline bikehire \
  --name sample \
  --s3-bucket skippr-public-sample-data \
  --s3-prefix bike-hire \
  --region us-east-1 \
  --format json
yaml
pipelines:
  bikehire:
    data_source: data_sources.sample

data_sources:
  sample:
    S3:
      s3_bucket: skippr-public-sample-data
      s3_prefix: bike-hire
      region: us-east-1
      format: json

A local file source is the same shape: DataSourceFile / connect data-source file / File: { path, format }.

Data sink ​

Leave the sink out and the WAL is the dataset. Add one when you want a warehouse.

python
from skippr import Config, DataSinkSnowflake, Pipeline

cfg = Config.discover()
warehouse = cfg.data_sink(
    "warehouse",
    DataSinkSnowflake(
        account="myorg-myaccount",
        user="skippr_loader",
        private_key_path="${SNOWFLAKE_PRIVATE_KEY_PATH}",
        warehouse="COMPUTE_WH",
        database="RAW_DATA",
        schema="PUBLIC",
        role="LOADER_ROLE",
        stage="@SKIPPR_STAGE",
    ),
)
cfg.pipeline("bikehire", Pipeline(data_source=cfg.get_data_source("sample"), data_sink=warehouse))
cfg.save()
bash
skipprd connect data-sink snowflake \
  --pipeline bikehire \
  --name warehouse \
  --account myorg-myaccount \
  --user skippr_loader \
  --private-key-path '${SNOWFLAKE_PRIVATE_KEY_PATH}' \
  --warehouse COMPUTE_WH \
  --database RAW_DATA \
  --schema PUBLIC \
  --role LOADER_ROLE \
  --stage '@SKIPPR_STAGE'
yaml
pipelines:
  bikehire:
    data_source: data_sources.sample
    data_sink: data_sinks.warehouse

data_sinks:
  warehouse:
    Snowflake:
      account: myorg-myaccount
      user: skippr_loader
      private_key_path: ${SNOWFLAKE_PRIVATE_KEY_PATH}
      warehouse: COMPUTE_WH
      database: RAW_DATA
      schema: PUBLIC
      role: LOADER_ROLE
      stage: "@SKIPPR_STAGE"

Postgres is DataSinkPostgres / connect data-sink postgres. File and S3 sinks are the same pattern.

Schema sink ​

Schema plugins live on the data sink entry (data_sinks.<name>.schema_sink), not on the pipeline. Athena uses Glue.

python
from skippr import Config, DataSinkAthena, Pipeline, SchemaSinkGlue

cfg = Config.discover()
glue = cfg.schema_sink(
    "glue",
    SchemaSinkGlue(
        s3_bucket="your-output-bucket",
        s3_prefix="data/bikehire",
        athena_workgroup_name="primary",
        athena_results_s3_bucket="your-athena-results",
        glue_database_name="skippr_quickstart",
        region="us-east-1",
        catalog="AwsDataCatalog",
    ),
)
lake = cfg.data_sink(
    "lake",
    DataSinkAthena(
        s3_bucket="your-output-bucket",
        s3_prefix="data/bikehire",
        athena_workgroup_name="primary",
        athena_results_s3_bucket="your-athena-results",
        glue_database_name="skippr_quickstart",
        region="us-east-1",
        catalog="AwsDataCatalog",
        format="parquet",
    ),
    schema_sink=glue,
)
cfg.pipeline("bikehire", Pipeline(data_source=cfg.get_data_source("sample"), data_sink=lake))
cfg.save()
bash
skipprd connect data-sink athena \
  --pipeline bikehire \
  --name lake \
  --s3-bucket your-output-bucket \
  --s3-prefix data/bikehire \
  --athena-workgroup-name primary \
  --athena-results-s3-bucket your-athena-results \
  --glue-database-name skippr_quickstart \
  --region us-east-1 \
  --catalog AwsDataCatalog \
  --format parquet

skipprd connect schema-sink glue \
  --pipeline bikehire \
  --name glue \
  --s3-bucket your-output-bucket \
  --s3-prefix data/bikehire \
  --athena-workgroup-name primary \
  --athena-results-s3-bucket your-athena-results \
  --glue-database-name skippr_quickstart \
  --region us-east-1 \
  --catalog AwsDataCatalog
yaml
pipelines:
  bikehire:
    data_source: data_sources.sample
    data_sink: data_sinks.lake

data_sinks:
  lake:
    Athena:
      s3_bucket: your-output-bucket
      s3_prefix: data/bikehire
      athena_workgroup_name: primary
      athena_results_s3_bucket: your-athena-results
      glue_database_name: skippr_quickstart
      region: us-east-1
      catalog: AwsDataCatalog
      format: parquet
    schema_sink: schema_sinks.glue

schema_sinks:
  glue:
    Glue:
      s3_bucket: your-output-bucket
      s3_prefix: data/bikehire
      athena_workgroup_name: primary
      athena_results_s3_bucket: your-athena-results
      glue_database_name: skippr_quickstart
      region: us-east-1
      catalog: AwsDataCatalog

Secrets ​

Plaintext secrets are rejected. Python secret fields take EnvRef("NAME"), which persists as ${NAME}. Quote ${ENV} in the shell so the shell does not expand it.

python
from skippr import Config, DataSinkPostgres, EnvRef, Pipeline

cfg = Config.discover()
postgres = cfg.data_sink(
    "postgres",
    DataSinkPostgres(
        host="localhost",
        user="skippr",
        password=EnvRef("POSTGRES_PASSWORD"),
        database="analytics",
    ),
)
cfg.pipeline("bikehire", Pipeline(data_source=cfg.get_data_source("sample"), data_sink=postgres))
cfg.save()
bash
skipprd connect data-sink postgres \
  --pipeline bikehire \
  --name postgres \
  --host localhost \
  --user skippr \
  --password '${POSTGRES_PASSWORD}' \
  --database analytics
yaml
pipelines:
  bikehire:
    data_source: data_sources.sample
    data_sink: data_sinks.postgres

data_sinks:
  postgres:
    Postgres:
      host: localhost
      user: skippr
      password: ${POSTGRES_PASSWORD}
      database: analytics

Then run it ​

python
import skippr

s = skippr.Session(skippr.Config.discover().get_pipeline("bikehire"))
s.doctor()
s.discover()
s.sync(once=True)
s.df()
bash
skipprd doctor
skipprd discover --pipeline bikehire --log
skipprd sync --pipeline bikehire --once --log
skipprd df --pipeline bikehire

Next ​

© 2026 Paul Hudson