slingdata-io/sling-cli/sling-python

uv add sling               # add to a uv project (pyproject.toml)
uv add 'sling[arrow]'      # + PyArrow: faster streaming, exact type preservation
uv pip install sling       # or into an existing venv
uv run --with sling script.py   # one-off run, no install step

from sling import Sling, Mode
# Database to database
Sling(
    src_conn="POSTGRES", src_stream="public.users",
    tgt_conn="SNOWFLAKE", tgt_object="public.users_copy",
    mode=Mode.FULL_REFRESH,
).run()
# SQL query to file
Sling(
    src_conn="POSTGRES",
    src_stream="select * from users where active = true",
    tgt_object="file:///tmp/active_users.csv",
).run()

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()

# Row dicts, memory-efficient iteration
for record in Sling(src_conn="POSTGRES", src_stream="public.users").stream():
    print(record["name"])
# Arrow: fastest, exact types (requires sling[arrow]); no tgt_object allowed
reader = Sling(src_conn="POSTGRES", src_stream="select * from big_table").stream_arrow()
for batch in reader:            # RecordBatchStreamReader — batch-wise for large data
    df = batch.to_pandas()
# or: table = reader.read_all()

from sling import Replication, ReplicationStream, Mode
# From a YAML file
Replication(file_path="path/to/replication.yaml").run()
# Built in code — kwargs mirror the replication YAML keys
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()

from sling import Pipeline
from sling.hooks import StepLog, StepCopy, StepReplication, StepHTTP, StepCommand
Pipeline(file_path="path/to/pipeline.yaml").run()
# Or typed steps (dicts with {"type": ...} also work)
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()

from sling import Connection
conn = Connection("POSTGRES")        # named conn or env-var URL both work
conn.test()                          # -> TestResult(success, error)
rows = conn.exec("select 1 as a")    # -> [{'a': 1}]
df = conn.exec("select * from users", return_type="dataframe", limit=0)

from sling.api_spec import ApiSpec
spec = ApiSpec.parse_file("path/to/spec.yaml")   # parse, modify, re-export
assert spec.validate() == []
spec.to_yaml_file("updated_spec.yaml")

@task
def load_orders():
    from sling import Replication
    Replication(file_path="replications/orders.yaml", env={"DATE": "{{ ds }}"}).run()