Dagster Integration:
Using Sling with Dagster

About this integration

This integration allows you to use Sling to extract and load data from popular data sources to destinations with high performance and ease.

Installation

pip install dagster-embedded-elt

Example

import dagster as dg
from dagster_embedded_elt.sling import (
    SlingConnectionResource,
    SlingResource,
    sling_assets,
)

source = SlingConnectionResource(
    name="MY_PG",
    type="postgres",
    host="localhost",
    port=5432,
    database="my_database",
    user="my_user",
    password=dg.EnvVar("PG_PASS"),
)

target = SlingConnectionResource(
    name="MY_SF",
    type="snowflake",
    host="hostname.snowflake",
    user="username",
    database="database",
    password=dg.EnvVar("SF_PASSWORD"),
    role="role",
)


@sling_assets(
    replication_config={
        "SOURCE": "MY_PG",
        "TARGET": "MY_SF",
        "defaults": {
            "mode": "full-refresh",
            "object": "{stream_schema}_{stream_table}",
        },
        "streams": {
            "public.accounts": None,
            "public.users": None,
            "public.finance_departments": {"object": "departments"},
        },
    }
)
def sling_assets(context, sling: SlingResource):
    yield from sling.replicate(context=context)


defs = dg.Definitions(
    assets=[sling_assets],
    resources={
        "sling": SlingResource(
            connections=[
                source,
                target,
            ]
        )
    },
)

About dlt

Sling provides an easy-to-use YAML configuration layer for loading data from files, replicating data between databases, exporting custom SQL queries to cloud storage, and much more.