embedded-elt (dagster-embedded-elt)¶

get_asset_key(stream_definition)[source]¶

A function that takes a stream definition from a Sling replication config and returns a Dagster AssetKey.

The stream definition is a dictionary key/value pair where the key is the stream name and the value is a dictionary representing the Sling Replication Stream Config.

For example:

stream_definition = {"public.users":
    {'sql': 'select all_user_id, name from public."all_Users"',
    'object': 'public.all_users'}
}

By default, this returns the class’s target_prefix paramater concatenated with the stream name. A stream named “public.accounts” will create an AssetKey named “target_public_accounts”.

Override this function to customize how to map a Sling stream to a Dagster AssetKey.

Alternatively, you can provide metadata in your Sling replication config to specify the Dagster AssetKey for a stream as follows:

public.users:
   meta:
     dagster:
       asset_key: "mydb_users"

Parameters:

stream_definition (Mapping[str, Any]) – A dictionary representing the stream definition

Returns:

The Dagster AssetKey for the replication stream.

Return type:

AssetKey

Examples

Using a custom mapping for streams:

class CustomSlingTranslator(DagsterSlingTranslator):
    @classmethod
    def get_asset_key_for_target(self, stream_definition) -> AssetKey:
        map = {"stream1": "asset1", "stream2": "asset2"}
        return AssetKey(map[stream_name])

classmethod get_auto_materialize_policy(stream_definition)[source]¶

classmethod get_deps_asset_key(stream_definition)[source]¶

A function that takes a stream name from a Sling replication config and returns a Dagster AssetKey for the dependencies of the replication stream.

By default, this returns the stream name. For example, a stream named “public.accounts” will create an AssetKey named “target_public_accounts” and a dependency named “public_accounts”.

Override this function to customize how to map a Sling stream to a Dagster depenency. Alternatively, you can provide metadata in your Sling replication config to specify the Dagster AssetKey for a stream as follows:

public.users:
    meta:
        dagster:
            deps: "sourcedb_users"

Parameters:

stream_name (str) – The name of the stream.

Returns:

The Dagster AssetKey dependency for the replication stream.

Return type:

AssetKey

Examples

Using a custom mapping for streams:

class CustomSlingTranslator(DagsterSlingTranslator):
    @classmethod
    def get_deps_asset_key(cls, stream_name: str) -> AssetKey:
        map = {"stream1": "asset1", "stream2": "asset2"}
        return AssetKey(map[stream_name])

classmethod get_description(stream_definition)[source]¶

classmethod get_freshness_policy(stream_definition)[source]¶

classmethod get_group_name(stream_definition)[source]¶

classmethod get_metadata(stream_definition)[source]¶

classmethod sanitize_stream_name(stream_name)[source]¶

A function that takes a stream name from a Sling replication config and returns a sanitized name for the stream.

By default, this removes any non-alphanumeric characters from the stream name and replaces them with underscores, while removing any double quotes.

Parameters:

stream_name (str) – The name of the stream.

Examples

Using a custom stream name sanitizer:

class CustomSlingTranslator(DagsterSlingTranslator):
    @classmethod
    def sanitize_stream_name(cls, stream_name: str) -> str:
        return stream_name.replace(".", "")

Resource for interacting with the Sling package. This resource can be used to run Sling replications.

Parameters:

• connections (List[SlingConnectionResource]) – A list of connections to use for the replication.

• source_connection (Optional[SlingSourceConnection]) – Deprecated, use connections instead.

• target_connection (Optional[SlingTargetConnection]) – Deprecated, use connections instead.

Examples

from dagster_etl.sling import SlingResource, SlingConnectionResource

sling_resource = SlingResource(
    connections=[
        SlingConnectionResource(
            name="MY_POSTGRES",
            type="postgres",
            connection_string=EnvVar("POSTGRES_CONNECTION_STRING"),
        ),
        SlingConnectionResource(
            name="MY_SNOWFLAKE",
            type="snowflake",
            host=EnvVar("SNOWFLAKE_HOST"),
            user=EnvVar("SNOWFLAKE_USER"),
            database=EnvVar("SNOWFLAKE_DATABASE"),
            password=EnvVar("SNOWFLAKE_PASSWORD"),
            role=EnvVar("SNOWFLAKE_ROLE"),
        ),
    ]
)

replicate(*, replication_config, dagster_sling_translator, debug=False)[source]¶

Runs a Sling replication from the given replication config.

Parameters:

• replication_config – The Sling replication config to use for the replication.

• dagster_sling_translator – The translator to use for the replication.

• debug – Whether to run the replication in debug mode.

Returns:

A generator of MaterializeResult

Return type:

Optional[Generator[MaterializeResult, None, None]]

A representation of a connection to a database or file to be used by Sling. This resource can be used as a source or a target for a Sling syncs.

Reference the Sling docs for more information on possible connection types and parameters: https://docs.slingdata.io/connections

The name of the connection is passed to Sling and must match the name of the connection provided in the replication configuration: https://docs.slingdata.io/sling-cli/run/configuration/replication You may provide either a connection string or keyword arguments for the connection.

Examples

Creating a Sling Connection for a file, such as CSV or JSON:

source = SlingConnectionResource(name="MY_FILE", type="file")

Create a Sling Connection for a Postgres database, using a connection string:

postgres_conn = SlingConnectionResource(name="MY_POSTGRES", type="postgres", connection_string=EnvVar("POSTGRES_CONNECTION_STRING"))
mysql_conn = SlingConnectionResource(name="MY_MYSQL", type="mysql", connection_string="mysql://user:password@host:port/schema")

Create a Sling Connection for a Postgres or Snowflake database, using keyword arguments: