Customizing components

info

This feature is considered in a preview stage and is under active development. There may be API changes and feature gaps. Please go to the #dg-components channel in our Slack to report issues or give feedback.

You can customize the behavior of a component beyond what is available in the component.yaml file.

To do so, you can create a subclass of your desired component in a file named component.py in the same directory as your component.yaml file.

from dagster_sling import SlingReplicationCollectionComponent


class CustomSubclass(SlingReplicationCollectionComponent): ...

You can then update the type: field in your component.yaml file to reference this new component type. It should be the fully qualified name of the type.

type: my_project.defs.my_def.CustomSubclass

attributes: ...

Customizing execution

For example, we can create a subclass of the SlingReplicationCollectionComponent that adds a debug log message during execution. SlingReplicationCollectionComponent has an execute method that can be overriden by subclasses.

from collections.abc import Iterator

from dagster_sling import SlingReplicationCollectionComponent, SlingResource

import dagster as dg


class DebugSlingReplicationComponent(SlingReplicationCollectionComponent):
    def execute(
        self, context: dg.AssetExecutionContext, sling: SlingResource
    ) -> Iterator:
        context.log.info("*******************CUSTOM*************************")
        return sling.replicate(context=context, debug=True)

Adding component-level templating scope

By default, the scopes available for use in the template are:

• env: A function that allows you to access environment variables.
• automation_condition: A scope allowing you to access all static constructors of the AutomationCondition class.

However, it can be useful to add additional scope options to your component type. For example, you may have a custom automation condition that you'd like to use in your component.

To do so, you can define a function that returns an AutomationCondition and define a get_additional_scope method on your subclass:

from collections.abc import Mapping
from typing import Any

from dagster_sling import SlingReplicationCollectionComponent

import dagster as dg


class SubclassWithScope(SlingReplicationCollectionComponent):
    @classmethod
    def get_additional_scope(cls) -> Mapping[str, Any]:
        def _custom_cron(cron_schedule: str) -> dg.AutomationCondition:
            return (
                dg.AutomationCondition.on_cron(cron_schedule)
                & ~dg.AutomationCondition.in_progress()
            )

        return {"custom_cron": _custom_cron}

This can then be used in your component.yaml file:

type: my_project.defs.my_def.CustomSubclass

attributes:
    ...
    asset_post_processors:
        - attributes:
            automation_condition: "{{ custom_cron('@daily') }}"