Skip to main content

Customizing components

warning

This feature is considered in a preview stage and is under active development. It can change significantly, or be removed completely. It is not considered ready for production use.

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_components.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_components.dagster_sling import SlingReplicationCollectionComponent
from dagster_sling import 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_components.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') }}"