| name | sling-python |
| description | Drive Sling from Python with the `sling` pip package — Sling, Replication, Pipeline and Connection classes, streaming records or DataFrames in/out, and orchestrator integration (Airflow, Dagster, scripts). Use when writing Python code that runs Sling, streams data into or out of Python, or builds replications/pipelines/API specs programmatically.
|
Sling Python Package
The sling pip package wraps the Sling CLI binary (auto-downloaded on first use). It adds Python-only capabilities: streaming records into/out of Python, DataFrame input, and programmatic config building.
Key principle: constructor kwargs mirror the YAML keys exactly. For what the keys mean (modes, options, hooks, selectors), load the concept skill — this skill only covers the Python invocation surface.
| Concept | Skill to load |
|---|
| Replication YAML semantics | sling-replications |
| Pipeline steps / hooks | sling-pipelines |
| API spec structure | sling-api-specs |
| Connection setup | CONNECTIONS.md |
Installation
Prefer uv — it's much faster and manages the venv for you:
uv add sling
uv add 'sling[arrow]'
uv pip install sling
uv run --with sling script.py
Plain pip also works: pip install sling / pip install 'sling[arrow]'. The sling binary auto-downloads on first use either way.
| Env Variable | Description |
|---|
SLING_BINARY | Path to a specific sling binary (e.g. a local dev build) |
SLING_PYTHON_USE_SHELL | Run the binary with shell=True (default false) |
Class Overview
| Class | Purpose |
|---|
Sling | Single task, mirrors CLI flags; supports input= and .stream() |
Replication / ReplicationStream | Multi-stream replication (file or built in code) |
Pipeline | Multi-step workflow (file or built in code) |
Connection | Test a connection, run SQL (list/dataframe/arrow results) |
Mode, Format, Compression, MergeStrategy | Enums for config values |
sling.hooks.Step* / Hook* | Typed pipeline steps and replication hooks |
Sling — single tasks and streaming
Kwargs mirror CLI flags: src_conn, src_stream, src_options, tgt_conn, tgt_object, tgt_options, mode, primary_key, update_key, select, where, limit, range, env, debug.
from sling import Sling, Mode
Sling(
src_conn="POSTGRES", src_stream="public.users",
tgt_conn="SNOWFLAKE", tgt_object="public.users_copy",
mode=Mode.FULL_REFRESH,
).run()
Sling(
src_conn="POSTGRES",
src_stream="select * from users where active = true",
tgt_object="file:///tmp/active_users.csv",
).run()
Input: Python data → target (input=)
Accepts list of dicts, a generator (memory-efficient), pandas or polars DataFrames (types preserved with sling[arrow]):
data = [{"id": 1, "name": "John"}, {"id": 2, "name": "Jane"}]
Sling(input=data, tgt_conn="POSTGRES", tgt_object="public.users").run()
import pandas as pd
df = pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]})
Sling(input=df, tgt_conn="POSTGRES", tgt_object="public.employees").run()
Output: source → Python (.stream() / .stream_arrow())
for record in Sling(src_conn="POSTGRES", src_stream="public.users").stream():
print(record["name"])
reader = Sling(src_conn="POSTGRES", src_stream="select * from big_table").stream_arrow()
for batch in reader:
df = batch.to_pandas()
Performance caveat: every Sling call spawns the binary and re-opens connections. For many tables in one run, use Replication (one process, one connection, parallel streams) instead of a Sling-per-table loop.
Replication — multi-stream, built dynamically
from sling import Replication, ReplicationStream, Mode
Replication(file_path="path/to/replication.yaml").run()
streams = {
folder: ReplicationStream(mode=Mode.FULL_REFRESH, object=table, primary_key="_hash_id")
for folder, table in folders
}
Replication(
source="AWS_S3", target="SNOWFLAKE",
defaults=ReplicationStream(mode=Mode.INCREMENTAL),
streams=streams,
env={"SLING_LOADED_AT_COLUMN": "true"},
).run()
Helpers: add_streams({...}), enable_streams([...]), disable_streams([...]), set_default_mode(mode).
Pipeline — multi-step workflows
from sling import Pipeline
from sling.hooks import StepLog, StepCopy, StepReplication, StepHTTP, StepCommand
Pipeline(file_path="path/to/pipeline.yaml").run()
Pipeline(
steps=[
StepCopy(from_="sftp/path/file.csv", to="aws_s3/path/file.csv"),
StepReplication(path="path/to/replication.yaml"),
StepHTTP(url="https://hooks.example.com/notify"),
StepCommand(command=["ls", "-l"], print_output=True),
StepLog(message="done"),
],
env={"MY_VAR": "value"},
).run()
Step classes (aliases of Hook*): StepQuery, StepHTTP, StepCheck, StepRead, StepWrite, StepCopy, StepDelete, StepLog, StepInspect, StepList, StepReplication, StepCommand, StepGroup, StepStore. See the sling-pipelines skill for each step's parameters.
Connection — test and query
from sling import Connection
conn = Connection("POSTGRES")
conn.test()
rows = conn.exec("select 1 as a")
df = conn.exec("select * from users", return_type="dataframe", limit=0)
return_type: list | dataframe | dataset | arrow (arrow streams, memory-bounded; others materialize fully).
limit=None applies the CLI default cap of 100 rows; pass limit=0 for no limit. For large results prefer Sling(...).stream().
ApiSpec — build API specs programmatically
sling.api_spec provides typed builders (ApiSpec, Endpoint, Request, Response, Pagination, Records, Processor, Rule, ...) that validate and serialize to spec YAML:
from sling.api_spec import ApiSpec
spec = ApiSpec.parse_file("path/to/spec.yaml")
assert spec.validate() == []
spec.to_yaml_file("updated_spec.yaml")
For spec structure (auth, pagination, processors), load the sling-api-specs skill.
Orchestrators (Airflow, Dagster, etc.)
Call Replication(...).run() or Pipeline(...).run() inside a task. Errors raise SlingError with the CLI output, so normal task retry/alerting applies. Pass per-run values via env= rather than editing YAML files.
@task
def load_orders():
from sling import Replication
Replication(file_path="replications/orders.yaml", env={"DATE": "{{ ds }}"}).run()
Full Documentation