Creating an inline component

Sometimes you need to define a component that you are unlikely to reuse. In this situation, you can create an inline component. An inline component is a component that is defined in the defs directory alongside its instance definitions, rather than in a directory exposed to the dg registry (for example, the components directory in a standard dg project).

Let's say we want to define a set of assets with each asset being materialized on its own schedule. This isn't a pattern we imagine using elsewhere in our project, so we define these asset/schedule pairs using an inline component.

To create the component, we use the dg scaffold inline-component command:

dg scaffold defs inline-component --typename AssetWithSchedule assets_with_schedules

Creating a Dagster inline component and corresponding component instance at assets_with_schedules.

Supposing we are in a project called my_project, this will create a new directory at my_project/defs/assets_with_schedules with two files:

• asset_with_schedule.py
• defs.yaml

assets_with_schedules.py

my_project/defs/assets_with_schedules/asset_with_schedule.py contains a basic definition of the AssetWithSchedule component class:

src/my_project/defs/assets_with_schedules/asset_with_schedule.py

import dagster as dg

class AssetWithSchedule(dg.Component, dg.Model, dg.Resolvable):
    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        return dg.Definitions()

Let's parameterize our component to take an asset key and a cron schedule string, and output a corresponding AssetsDefinition and ScheduleDefinition. Because this is just for illustrative purposes, we'll just set the asset materialization function to generate random numbers:

src/my_project/defs/assets_with_schedules/asset_with_schedule.py

from random import randint

import dagster as dg


class AssetWithSchedule(dg.Component, dg.Model, dg.Resolvable):
    asset_key: list[str]
    cron_schedule: str

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        @dg.asset(key=dg.AssetKey(self.asset_key))
        def asset():
            return randint(1, 100)

        schedule = dg.ScheduleDefinition(
            name=f"{'_'.join(self.asset_key)}_schedule",
            cron_schedule=self.cron_schedule,
            target=asset,
        )

        return dg.Definitions(assets=[asset], schedules=[schedule])

defs.yaml

my_project/defs/assets_with_schedules/defs.yaml starts us off with an instance of the new component with no defined attributes:

src/my_project/defs/assets_with_schedules/defs.yaml

type: my_project.defs.assets_with_schedules.asset_with_schedule.AssetWithSchedule
attributes: {}

Note that there is nothing special about the type field. It is just a fully-qualified Python object reference, as it is when defining instances of reusable registered components. The only difference between AssetWithSchedule and a registered component is that AssetWithSchedule won't show up in the dg registry.

We'll modify this file to define three instances of our component, each representing an asset with a corresponding schedule:

src/my_project/defs/assets_with_schedules/defs.yaml

type: my_project.defs.assets_with_schedules.asset_with_schedule.AssetWithSchedule
attributes:
  asset_key: ["foo"]
  cron_schedule: "*/10 * * * *"
---
type: my_project.defs.assets_with_schedules.asset_with_schedule.AssetWithSchedule
attributes:
  asset_key: ["bar"]
  cron_schedule: "*/20 * * * *"
---
type: my_project.defs.assets_with_schedules.asset_with_schedule.AssetWithSchedule
attributes:
  asset_key: ["baz"]
  cron_schedule: "*/30 * * * *"

If we now run dg list defs, we can see our three assets and schedules:

dg list defs

┏━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Section   ┃ Definitions                                    ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ Assets    │ ┏━━━━━┳━━━━━━━━━┳━━━━━━┳━━━━━━━┳━━━━━━━━━━━━━┓ │
│           │ ┃ Key ┃ Group   ┃ Deps ┃ Kinds ┃ Description ┃ │
│           │ ┡━━━━━╇━━━━━━━━━╇━━━━━━╇━━━━━━━╇━━━━━━━━━━━━━┩ │
│           │ │ bar │ default │      │       │             │ │
│           │ ├─────┼─────────┼──────┼───────┼─────────────┤ │
│           │ │ baz │ default │      │       │             │ │
│           │ ├─────┼─────────┼──────┼───────┼─────────────┤ │
│           │ │ foo │ default │      │       │             │ │
│           │ └─────┴─────────┴──────┴───────┴─────────────┘ │
│ Schedules │ ┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┓                │
│           │ ┃ Key          ┃ Cron         ┃                │
│           │ ┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━┩                │
│           │ │ bar_schedule │ */20 * * * * │                │
│           │ ├──────────────┼──────────────┤                │
│           │ │ baz_schedule │ */30 * * * * │                │
│           │ ├──────────────┼──────────────┤                │
│           │ │ foo_schedule │ */10 * * * * │                │
│           │ └──────────────┴──────────────┘                │
└───────────┴────────────────────────────────────────────────┘

Now if we run dg dev and look at the "Automations" tab, we should see three schedules targeting their corresponding assets: