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. This subclass should be annotated with the @component_type decorator, which will define a local name for this component:

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. The new type name will be .<component-name>, where the leading . indicates that this is a local component type:

type: .custom_subclass

attributes:
    ...

Customizing execution​

By convention, most library components have an execute() method that defines the core runtime behavior of the component. This can be overridden by subclasses of the component to customize this behavior.

For example, we can create a subclass of the SlingReplicationCollectionComponent that adds a debug log message during execution:

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:

component_type: .custom_subclass

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