Customizing components

info

This feature is still in development and might change in patch releases. It’s not production ready, and the documentation may also evolve. Stay tuned for updates.

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 import registered_component_type
from dagster_components.lib import SlingReplicationCollection


@registered_component_type(name="custom_subclass")
class CustomSubclass(SlingReplicationCollection): ...

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 import registered_component_type
from dagster_components.lib import SlingReplicationCollection
from dagster_sling import SlingResource

import dagster as dg


@registered_component_type(name="debug_sling_replication")
class DebugSlingReplicationComponent(SlingReplicationCollection):
    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 import registered_component_type
from dagster_components.lib import SlingReplicationCollection

import dagster as dg


@registered_component_type(name="custom_subclass")
class SubclassWithScope(SlingReplicationCollection):
    def get_additional_scope(self) -> 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') }}"