Connect

connect writes skippr.yml. Python Session.connect(), 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. Existing skippr.yml or skippr.yaml is reused. The write is read → parse → merge → write, so sibling pipelines and extra keys on the same plugin survive.

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

Workspace

Python

import skipprd
from skipprd import StorageMode

skipprd.workspace("quickstart").storage_mode(StorageMode.LOCAL)

CLI

skipprd --workspace quickstart --storage-mode local connect data-source s3 \
  --pipeline bikehire \
  --name sample \
  --s3-bucket skippr-public-sample-data \
  --s3-prefix bike-hire

skippr.yml

skippr:
  workspace: quickstart
  skipprd_el_storage_mode: local

--storage-mode is local or s3. --offset-store is sled, dynamodb, or cloud-tables. WAL backend stays --wal-storage / WAL_STORAGE (not a YAML key).

Python writes when required fields are set. Secret fields must be ${ENV} references.

Data source

Python

import skipprd
from skipprd import DataSource

s = skipprd.Session(pipeline="bikehire")
(
    s.connect()
    .data_source(DataSource.S3)
    .name("sample")
    .s3_bucket("skippr-public-sample-data")
    .s3_prefix("bike-hire")
    .region("us-east-1")
    .format("json")
)

CLI

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

skippr.yml

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: DataSource.File / 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

import skipprd
from skipprd import DataSink

s = skipprd.Session(pipeline="bikehire")
(
    s.connect()
    .data_sink(DataSink.Snowflake)
    .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")
)

CLI

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'

skippr.yml

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 DataSink.Postgres / 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

import skipprd
from skipprd import DataSink, SchemaSink

s = skipprd.Session(pipeline="bikehire")
(
    s.connect()
    .data_sink(DataSink.Athena)
    .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")
)
(
    s.connect()
    .schema_sink(SchemaSink.Glue)
    .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")
)

CLI

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

skippr.yml

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. Quote ${ENV} in the shell so the shell does not expand it.

skipprd connect data-sink postgres \
  --pipeline bikehire \
  --name warehouse \
  --host localhost \
  --user skippr \
  --password '${POSTGRES_PASSWORD}' \
  --database analytics

Then run it

Python

import skipprd

s = skipprd.Session(config="skippr.yml", pipeline="bikehire")
s.doctor()
s.discover()
s.sync(once=True)
s.df()

CLI

skipprd doctor
skipprd discover --pipeline bikehire --log
skipprd sync --pipeline bikehire --once --log
skipprd df --pipeline bikehire

Next

• Getting started — public sample, no warehouse
• Python — Session, query()
• Snowflake and Postgres