Ask AI

Schedules and sensors

Dagster offers several ways to run data pipelines without manual intervation, including traditional scheduling and event-based triggers. Automating your Dagster pipelines can boost efficiency and ensure that data is produced consistently and reliably.


Run requests

class dagster.RunRequest(run_key=None, run_config=None, tags=None, job_name=None, asset_selection=None, stale_assets_only=False, partition_key=None, asset_check_keys=None, **kwargs)[source]
class dagster.SkipReason(skip_message=None)[source]

Represents a skipped evaluation, where no runs are requested. May contain a message to indicate why no runs were requested.

skip_message

A message displayed in the Dagster UI for why this evaluation resulted in no requested runs.

Type:

Optional[str]


Schedules

Schedules are Dagster’s way to support traditional ways of automation, such as specifying a job should run at Mondays at 9:00AM. Jobs triggered by schedules can contain a subset of assets or ops.

@dagster.schedule(cron_schedule, *, job_name=None, name=None, tags=None, tags_fn=None, metadata=None, should_execute=None, environment_vars=None, execution_timezone=None, description=None, job=None, default_status=DefaultScheduleStatus.STOPPED, required_resource_keys=None, target=None)[source]

Creates a schedule following the provided cron schedule and requests runs for the provided job.

The decorated function takes in a ScheduleEvaluationContext as its only argument, and does one of the following:

  1. Return a RunRequest object.

  2. Return a list of RunRequest objects.

  3. Return a SkipReason object, providing a descriptive message of why no runs were requested.

  4. Return nothing (skipping without providing a reason)

  5. Return a run config dictionary.

  6. Yield a SkipReason or yield one ore more RunRequest objects.

Returns a ScheduleDefinition.

Parameters:
  • cron_schedule (Union[str, Sequence[str]]) – A valid cron string or sequence of cron strings specifying when the schedule will run, e.g., 45 23 * * 6 for a schedule that runs at 11:45 PM every Saturday. If a sequence is provided, then the schedule will run for the union of all execution times for the provided cron strings, e.g., ['45 23 * * 6', '30 9 * * 0'] for a schedule that runs at 11:45 PM every Saturday and 9:30 AM every Sunday.

  • name (Optional[str]) – The name of the schedule.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the schedule and can be used for searching and filtering in the UI.

  • tags_fn (Optional[Callable[[ScheduleEvaluationContext], Optional[Dict[str, str]]]]) – A function that generates tags to attach to the schedule’s runs. Takes a ScheduleEvaluationContext and returns a dictionary of tags (string key-value pairs). Note: Either tags or tags_fn may be set, but not both.

  • metadata (Optional[Mapping[str, Any]]) – A set of metadata entries that annotate the schedule. Values will be normalized to typed MetadataValue objects.

  • should_execute (Optional[Callable[[ScheduleEvaluationContext], bool]]) – A function that runs at schedule execution time to determine whether a schedule should execute or skip. Takes a ScheduleEvaluationContext and returns a boolean (True if the schedule should execute). Defaults to a function that always returns True.

  • execution_timezone (Optional[str]) – Timezone in which the schedule should run. Supported strings for timezones are the ones provided by the IANA time zone database - e.g. "America/Los_Angeles".

  • description (Optional[str]) – A human-readable description of the schedule.

  • job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job that should execute when the schedule runs.

  • default_status (DefaultScheduleStatus) – If set to RUNNING, the schedule will immediately be active when starting Dagster. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • required_resource_keys (Optional[Set[str]]) – The set of resource keys required by the schedule.

  • target (Optional[Union[CoercibleToAssetSelection, AssetsDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – experimental (This parameter may break in future versions, even between dot releases.) The target that the schedule will execute. It can take AssetSelection objects and anything coercible to it (e.g. str, Sequence[str], AssetKey, AssetsDefinition). It can also accept JobDefinition (a function decorated with @job is an instance of JobDefinition) and UnresolvedAssetJobDefinition (the return value of define_asset_job()) objects. This is an experimental parameter that will replace job and job_name.

class dagster.ScheduleDefinition(name=None, *, cron_schedule=None, job_name=None, run_config=None, run_config_fn=None, tags=None, tags_fn=None, metadata=None, should_execute=None, environment_vars=None, execution_timezone=None, execution_fn=None, description=None, job=None, default_status=DefaultScheduleStatus.STOPPED, required_resource_keys=None, target=None)[source]

Defines a schedule that targets a job.

Parameters:
  • name (Optional[str]) – The name of the schedule to create. Defaults to the job name plus _schedule.

  • cron_schedule (Union[str, Sequence[str]]) – A valid cron string or sequence of cron strings specifying when the schedule will run, e.g., 45 23 * * 6 for a schedule that runs at 11:45 PM every Saturday. If a sequence is provided, then the schedule will run for the union of all execution times for the provided cron strings, e.g., ['45 23 * * 6', '30 9 * * 0] for a schedule that runs at 11:45 PM every Saturday and 9:30 AM every Sunday.

  • execution_fn (Callable[ScheduleEvaluationContext]) –

    The core evaluation function for the schedule, which is run at an interval to determine whether a run should be launched or not. Takes a ScheduleEvaluationContext.

    This function must return a generator, which must yield either a single SkipReason or one or more RunRequest objects.

  • run_config (Optional[Union[RunConfig, Mapping]]) – The config that parameterizes this execution, as a dict.

  • run_config_fn (Optional[Callable[[ScheduleEvaluationContext], [Mapping]]]) – A function that takes a ScheduleEvaluationContext object and returns the run configuration that parameterizes this execution, as a dict. Note: Only one of the following may be set: You may set run_config, run_config_fn, or execution_fn.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the schedule and can be used for searching and filtering in the UI. If no execution_fn is provided, then these will also be automatically attached to runs launched by the schedule.

  • tags_fn (Optional[Callable[[ScheduleEvaluationContext], Optional[Mapping[str, str]]]]) – A function that generates tags to attach to the schedule’s runs. Takes a ScheduleEvaluationContext and returns a dictionary of tags (string key-value pairs). Note: Only one of the following may be set: tags, tags_fn, or execution_fn.

  • should_execute (Optional[Callable[[ScheduleEvaluationContext], bool]]) – A function that runs at schedule execution time to determine whether a schedule should execute or skip. Takes a ScheduleEvaluationContext and returns a boolean (True if the schedule should execute). Defaults to a function that always returns True.

  • execution_timezone (Optional[str]) –

    Timezone in which the schedule should run. Supported strings for timezones are the ones provided by the IANA time zone database - e.g. "America/Los_Angeles".

  • description (Optional[str]) – A human-readable description of the schedule.

  • job (Optional[Union[GraphDefinition, JobDefinition]]) – The job that should execute when this schedule runs.

  • default_status (DefaultScheduleStatus) – If set to RUNNING, the schedule will start as running. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • required_resource_keys (Optional[Set[str]]) – The set of resource keys required by the schedule.

  • target (Optional[Union[CoercibleToAssetSelection, AssetsDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – experimental (This parameter may break in future versions, even between dot releases.) The target that the schedule will execute. It can take AssetSelection objects and anything coercible to it (e.g. str, Sequence[str], AssetKey, AssetsDefinition). It can also accept JobDefinition (a function decorated with @job is an instance of JobDefinition) and UnresolvedAssetJobDefinition (the return value of define_asset_job()) objects. This is an experimental parameter that will replace job and job_name.

  • metadata (Optional[Mapping[str, Any]]) – A set of metadata entries that annotate the schedule. Values will be normalized to typed MetadataValue objects. Not currently shown in the UI but available at runtime via ScheduleEvaluationContext.repository_def.get_schedule_def(<name>).metadata.

property cron_schedule

The cron schedule representing when this schedule will be evaluated.

Type:

Union[str, Sequence[str]]

property default_status

The default status for this schedule when it is first loaded in a code location.

Type:

DefaultScheduleStatus

property description

A description for this schedule.

Type:

Optional[str]

property environment_vars

deprecated This API will be removed in version 2.0.

Setting this property no longer has any effect..

Environment variables to export to the cron schedule.

Type:

Mapping[str, str]

property execution_timezone

The timezone in which this schedule will be evaluated.

Type:

Optional[str]

property job

The job that is targeted by this schedule.

Type:

Union[JobDefinition, UnresolvedAssetJobDefinition]

property job_name

The name of the job targeted by this schedule.

Type:

str

property metadata

The metadata for this schedule.

Type:

Mapping[str, str]

property name

The name of the schedule.

Type:

str

property required_resource_keys

The set of keys for resources that must be provided to this schedule.

Type:

Set[str]

property tags

The tags for this schedule.

Type:

Mapping[str, str]

class dagster.ScheduleEvaluationContext(instance_ref, scheduled_execution_time, log_key=None, repository_name=None, schedule_name=None, resources=None, repository_def=None)[source]

The context object available as the first argument to various functions defined on a dagster.ScheduleDefinition.

A ScheduleEvaluationContext object is passed as the first argument to run_config_fn, tags_fn, and should_execute.

Users should not instantiate this object directly. To construct a ScheduleEvaluationContext for testing purposes, use dagster.build_schedule_context().

Example

from dagster import schedule, ScheduleEvaluationContext

@schedule
def the_schedule(context: ScheduleEvaluationContext):
    ...
property instance

The current DagsterInstance.

Type:

DagsterInstance

property resources

Mapping of resource key to resource definition to be made available during schedule execution.

property scheduled_execution_time

The time in which the execution was scheduled to happen. May differ slightly from both the actual execution time and the time at which the run config is computed.

dagster.build_schedule_context(instance=None, scheduled_execution_time=None, resources=None, repository_def=None, instance_ref=None)[source]

Builds schedule execution context using the provided parameters.

The instance provided to build_schedule_context must be persistent; DagsterInstance.ephemeral() will result in an error.

Parameters:
  • instance (Optional[DagsterInstance]) – The Dagster instance configured to run the schedule.

  • scheduled_execution_time (datetime) – The time in which the execution was scheduled to happen. May differ slightly from both the actual execution time and the time at which the run config is computed.

Examples

context = build_schedule_context(instance)
dagster.build_schedule_from_partitioned_job(job, description=None, name=None, minute_of_hour=None, hour_of_day=None, day_of_week=None, day_of_month=None, default_status=DefaultScheduleStatus.STOPPED, tags=None, cron_schedule=None, execution_timezone=None)[source]

Creates a schedule from a job that targets time window-partitioned or statically-partitioned assets. The job can also be multi-partitioned, as long as one of the partition dimensions is time-partitioned.

The schedule executes at the cadence specified by the time partitioning of the job or assets.

Example:
######################################
# Job that targets partitioned assets
######################################

from dagster import (
    DailyPartitionsDefinition,
    asset,
    build_schedule_from_partitioned_job,
    define_asset_job,
    Definitions,
)

@asset(partitions_def=DailyPartitionsDefinition(start_date="2020-01-01"))
def asset1():
    ...

asset1_job = define_asset_job("asset1_job", selection=[asset1])

# The created schedule will fire daily
asset1_job_schedule = build_schedule_from_partitioned_job(asset1_job)

defs = Definitions(assets=[asset1], schedules=[asset1_job_schedule])

################
# Non-asset job
################

from dagster import DailyPartitionsDefinition, build_schedule_from_partitioned_job, jog


@job(partitions_def=DailyPartitionsDefinition(start_date="2020-01-01"))
def do_stuff_partitioned():
    ...

# The created schedule will fire daily
do_stuff_partitioned_schedule = build_schedule_from_partitioned_job(
    do_stuff_partitioned,
)

defs = Definitions(schedules=[do_stuff_partitioned_schedule])
dagster._core.scheduler.DagsterDaemonScheduler Scheduler[source]

Config Schema:
max_catchup_runs (dagster.IntSource, optional):

For partitioned schedules, controls the maximum number of past partitions for each schedule that will be considered when looking for missing runs . Generally this parameter will only come into play if the scheduler falls behind or launches after experiencing downtime. This parameter will not be checked for schedules without partition sets (for example, schedules created using the @schedule decorator) - only the most recent execution time will be considered for those schedules.

Note: No matter what this value is, the scheduler will never launch a run from a time before the schedule was turned on, even if the schedule’s start_date is earlier. If you want to launch runs for earlier partitions, launch a backfill.

Default Value: 5

max_tick_retries (dagster.IntSource, optional):

For each schedule tick that raises an error, the number of times to retry the tick.

Default Value: 0

Default scheduler implementation that submits runs from the long-lived dagster-daemon process. Periodically checks each running schedule for execution times that don’t yet have runs and launches them.


Sensors

Sensors are typically used to poll, listen, and respond to external events. For example, you could configure a sensor to run a job or materialize an asset in response to specific events.

@dagster.sensor(job_name=None, *, name=None, minimum_interval_seconds=None, description=None, job=None, jobs=None, default_status=DefaultSensorStatus.STOPPED, asset_selection=None, required_resource_keys=None, tags=None, metadata=None, target=None)[source]

Creates a sensor where the decorated function is used as the sensor’s evaluation function.

The decorated function may:

  1. Return a RunRequest object.

  2. Return a list of RunRequest objects.

  3. Return a SkipReason object, providing a descriptive message of why no runs were requested.

  4. Return nothing (skipping without providing a reason)

  5. Yield a SkipReason or yield one or more RunRequest objects.

Takes a SensorEvaluationContext.

Parameters:
  • name (Optional[str]) – The name of the sensor. Defaults to the name of the decorated function.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job to be executed when the sensor fires.

  • jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]]) – (experimental) A list of jobs to be executed when the sensor fires.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • asset_selection (Optional[Union[str, Sequence[str], Sequence[AssetKey], Sequence[Union[AssetsDefinition, SourceAsset]], AssetSelection]]) – (Experimental) an asset selection to launch a run for if the sensor condition is met. This can be provided instead of specifying a job.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

  • target (Optional[Union[CoercibleToAssetSelection, AssetsDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – experimental (This parameter may break in future versions, even between dot releases.) The target that the sensor will execute. It can take AssetSelection objects and anything coercible to it (e.g. str, Sequence[str], AssetKey, AssetsDefinition). It can also accept JobDefinition (a function decorated with @job is an instance of JobDefinition) and UnresolvedAssetJobDefinition (the return value of define_asset_job()) objects. This is an experimental parameter that will replace job, jobs, and asset_selection.

class dagster.SensorDefinition(name=None, *, evaluation_fn=None, job_name=None, minimum_interval_seconds=None, description=None, job=None, jobs=None, default_status=DefaultSensorStatus.STOPPED, asset_selection=None, required_resource_keys=None, tags=None, metadata=None, target=None)[source]

Define a sensor that initiates a set of runs based on some external state.

Parameters:
  • evaluation_fn (Callable[[SensorEvaluationContext]]) –

    The core evaluation function for the sensor, which is run at an interval to determine whether a run should be launched or not. Takes a SensorEvaluationContext.

    This function must return a generator, which must yield either a single SkipReason or one or more RunRequest objects.

  • name (Optional[str]) – The name of the sensor to create. Defaults to name of evaluation_fn

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • job (Optional[GraphDefinition, JobDefinition, UnresolvedAssetJob]) – The job to execute when this sensor fires.

  • jobs (Optional[Sequence[GraphDefinition, JobDefinition, UnresolvedAssetJob]]) – (experimental) A list of jobs to execute when this sensor fires.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • asset_selection (Optional[Union[str, Sequence[str], Sequence[AssetKey], Sequence[Union[AssetsDefinition, SourceAsset]], AssetSelection]]) – (Experimental) an asset selection to launch a run for if the sensor condition is met. This can be provided instead of specifying a job.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects. Not currently shown in the UI but available at runtime via SensorEvaluationContext.repository_def.get_sensor_def(<name>).metadata.

  • target (Optional[Union[CoercibleToAssetSelection, AssetsDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – experimental (This parameter may break in future versions, even between dot releases.) The target that the sensor will execute. It can take AssetSelection objects and anything coercible to it (e.g. str, Sequence[str], AssetKey, AssetsDefinition). It can also accept JobDefinition (a function decorated with @job is an instance of JobDefinition) and UnresolvedAssetJobDefinition (the return value of define_asset_job()) objects. This is an experimental parameter that will replace job, jobs, and asset_selection.

property default_status

The default status for this sensor when it is first loaded in a code location.

Type:

DefaultSensorStatus

property description

A description for this sensor.

Type:

Optional[str]

property job

The job that is targeted by this schedule.

Type:

Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]

property job_name

The name of the job that is targeted by this sensor.

Type:

Optional[str]

property jobs

A list of jobs that are targeted by this schedule.

Type:

List[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]

property minimum_interval_seconds

The minimum number of seconds between sequential evaluations of this sensor.

Type:

Optional[int]

property name

The name of this sensor.

Type:

str

property required_resource_keys

The set of keys for resources that must be provided to this sensor.

Type:

Set[str]

class dagster.SensorEvaluationContext(instance_ref, last_tick_completion_time=None, last_run_key=None, cursor=None, log_key=None, repository_name=None, repository_def=None, instance=None, sensor_name=None, resources=None, definitions=None, last_sensor_start_time=None, code_location_origin=None, last_completion_time=None)[source]

The context object available as the argument to the evaluation function of a dagster.SensorDefinition.

Users should not instantiate this object directly. To construct a SensorEvaluationContext for testing purposes, use dagster. build_sensor_context().

instance_ref

The serialized instance configured to run the schedule

Type:

Optional[InstanceRef]

cursor

The cursor, passed back from the last sensor evaluation via the cursor attribute of SkipReason and RunRequest

Type:

Optional[str]

last_tick_completion_time

The last time that the sensor was evaluated (UTC).

Type:

float

last_run_key

DEPRECATED The run key of the RunRequest most recently created by this sensor. Use the preferred cursor attribute instead.

Type:

str

log_key

The log key to use for this sensor tick.

Type:

Optional[List[str]]

repository_name

The name of the repository that the sensor belongs to.

Type:

Optional[str]

repository_def

The repository or that the sensor belongs to. If needed by the sensor top-level resource definitions will be pulled from this repository. You can provide either this or definitions.

Type:

Optional[RepositoryDefinition]

instance

The deserialized instance can also be passed in directly (primarily useful in testing contexts).

Type:

Optional[DagsterInstance]

definitions

Definitions object that the sensor is defined in. If needed by the sensor, top-level resource definitions will be pulled from these definitions. You can provide either this or repository_def.

Type:

Optional[Definitions]

resources

A dict of resource keys to resource definitions to be made available during sensor execution.

Type:

Optional[Dict[str, Any]]

last_sensor_start_time

The last time that the sensor was started (UTC).

Type:

float

code_location_origin

The code location that the sensor is in.

Type:

Optional[CodeLocationOrigin]

Example

from dagster import sensor, SensorEvaluationContext

@sensor
def the_sensor(context: SensorEvaluationContext):
    ...
update_cursor(cursor)[source]

Updates the cursor value for this sensor, which will be provided on the context for the next sensor evaluation.

This can be used to keep track of progress and avoid duplicate work across sensor evaluations.

Parameters:

cursor (Optional[str])

property cursor

The cursor value for this sensor, which was set in an earlier sensor evaluation.

property instance

The current DagsterInstance.

Type:

DagsterInstance

property is_first_tick_since_sensor_start

Flag representing if this is the first tick since the sensor was started.

property last_run_key

The run key supplied to the most recent RunRequest produced by this sensor.

Type:

Optional[str]

property last_sensor_start_time

Timestamp representing the last time this sensor was started. Can be used in concert with last_tick_completion_time to determine if this is the first tick since the sensor was started.

Type:

Optional[float]

property last_tick_completion_time

Timestamp representing the last time this sensor completed an evaluation.

Type:

Optional[float]

property repository_def

The RepositoryDefinition that this sensor resides in.

Type:

Optional[RepositoryDefinition]

property repository_name

The name of the repository that this sensor resides in.

Type:

Optional[str]

property resources

A mapping from resource key to instantiated resources for this sensor.

Type:

Resources

dagster.build_sensor_context(instance=None, cursor=None, repository_name=None, repository_def=None, sensor_name=None, resources=None, definitions=None, instance_ref=None, last_sensor_start_time=None)[source]

Builds sensor execution context using the provided parameters.

This function can be used to provide a context to the invocation of a sensor definition.If provided, the dagster instance must be persistent; DagsterInstance.ephemeral() will result in an error.

Parameters:
  • instance (Optional[DagsterInstance]) – The dagster instance configured to run the sensor.

  • cursor (Optional[str]) – A cursor value to provide to the evaluation of the sensor.

  • repository_name (Optional[str]) – The name of the repository that the sensor belongs to.

  • repository_def (Optional[RepositoryDefinition]) – The repository that the sensor belongs to. If needed by the sensor top-level resource definitions will be pulled from this repository. You can provide either this or definitions.

  • resources (Optional[Mapping[str, ResourceDefinition]]) – A set of resource definitions to provide to the sensor. If passed, these will override any resource definitions provided by the repository.

  • definitions (Optional[Definitions]) – Definitions object that the sensor is defined in. If needed by the sensor, top-level resource definitions will be pulled from these definitions. You can provide either this or repository_def.

  • last_sensor_start_time (Optional[float]) – The last time the sensor was started.

Examples

context = build_sensor_context()
my_sensor(context)
@dagster.asset_sensor(asset_key, *, job_name=None, name=None, minimum_interval_seconds=None, description=None, job=None, jobs=None, default_status=DefaultSensorStatus.STOPPED, required_resource_keys=None, tags=None, metadata=None)[source]

Creates an asset sensor where the decorated function is used as the asset sensor’s evaluation function.

If the asset has been materialized multiple times between since the last sensor tick, the evaluation function will only be invoked once, with the latest materialization.

The decorated function may:

  1. Return a RunRequest object.

  2. Return a list of RunRequest objects.

  3. Return a SkipReason object, providing a descriptive message of why no runs were requested.

  4. Return nothing (skipping without providing a reason)

  5. Yield a SkipReason or yield one or more RunRequest objects.

Takes a SensorEvaluationContext and an EventLogEntry corresponding to an AssetMaterialization event.

Parameters:
  • asset_key (AssetKey) – The asset_key this sensor monitors.

  • name (Optional[str]) – The name of the sensor. Defaults to the name of the decorated function.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job to be executed when the sensor fires.

  • jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]]) – (experimental) A list of jobs to be executed when the sensor fires.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI. Values that are not already strings will be serialized as JSON.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

Example

from dagster import AssetKey, EventLogEntry, SensorEvaluationContext, asset_sensor


@asset_sensor(asset_key=AssetKey("my_table"), job=my_job)
def my_asset_sensor(context: SensorEvaluationContext, asset_event: EventLogEntry):
    return RunRequest(
        run_key=context.cursor,
        run_config={
            "ops": {
                "read_materialization": {
                    "config": {
                        "asset_key": asset_event.dagster_event.asset_key.path,
                    }
                }
            }
        },
    )
class dagster.AssetSensorDefinition(name, asset_key, job_name, asset_materialization_fn, minimum_interval_seconds=None, description=None, job=None, jobs=None, default_status=DefaultSensorStatus.STOPPED, required_resource_keys=None, tags=None, metadata=None)[source]

Define an asset sensor that initiates a set of runs based on the materialization of a given asset.

If the asset has been materialized multiple times between since the last sensor tick, the evaluation function will only be invoked once, with the latest materialization.

Parameters:
  • name (str) – The name of the sensor to create.

  • asset_key (AssetKey) – The asset_key this sensor monitors.

  • asset_materialization_fn (Callable[[SensorEvaluationContext, EventLogEntry], Union[Iterator[Union[RunRequest, SkipReason]], RunRequest, SkipReason]]) –

    The core evaluation function for the sensor, which is run at an interval to determine whether a run should be launched or not. Takes a SensorEvaluationContext and an EventLogEntry corresponding to an AssetMaterialization event.

    This function must return a generator, which must yield either a single SkipReason or one or more RunRequest objects.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job object to target with this sensor.

  • jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]]) – (experimental) A list of jobs to be executed when the sensor fires.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

property asset_key

The key of the asset targeted by this sensor.

Type:

AssetKey

@dagster.freshness_policy_sensor(asset_selection, *, name=None, minimum_interval_seconds=None, description=None, default_status=DefaultSensorStatus.STOPPED)[source]

experimental This API may break in future versions, even between dot releases.

Define a sensor that reacts to the status of a given set of asset freshness policies, where the decorated function will be evaluated on every tick for each asset in the selection that has a FreshnessPolicy defined.

Note: returning or yielding a value from the annotated function will result in an error.

Takes a FreshnessPolicySensorContext.

Parameters:
  • asset_selection (AssetSelection) – The asset selection monitored by the sensor.

  • name (Optional[str]) – The name of the sensor. Defaults to the name of the decorated function.

  • freshness_policy_sensor_fn (Callable[[FreshnessPolicySensorContext], None]) – The core evaluation function for the sensor. Takes a FreshnessPolicySensorContext.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

class dagster.FreshnessPolicySensorDefinition(name, asset_selection, freshness_policy_sensor_fn, minimum_interval_seconds=None, description=None, default_status=DefaultSensorStatus.STOPPED, required_resource_keys=None)[source]

Define a sensor that reacts to the status of a given set of asset freshness policies, where the decorated function will be evaluated on every sensor tick.

Parameters:
  • name (str) – The name of the sensor. Defaults to the name of the decorated function.

  • freshness_policy_sensor_fn (Callable[[FreshnessPolicySensorContext], None]) – The core evaluation function for the sensor. Takes a FreshnessPolicySensorContext.

  • asset_selection (AssetSelection) – The asset selection monitored by the sensor.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

class dagster.FreshnessPolicySensorContext(sensor_name, asset_key, freshness_policy, minutes_overdue, previous_minutes_overdue, instance, resources=None)[source]

The context object available to a decorated function of freshness_policy_sensor.

sensor_name

the name of the sensor.

Type:

str

asset_key

the key of the asset being monitored

Type:

AssetKey

freshness_policy

the freshness policy of the asset being monitored

Type:

FreshnessPolicy

minutes_overdue
Type:

Optional[float]

previous_minutes_overdue

the minutes_overdue value for this asset on the previous sensor tick.

Type:

Optional[float]

instance

the current instance.

Type:

DagsterInstance

dagster.build_freshness_policy_sensor_context(sensor_name, asset_key, freshness_policy, minutes_overdue, previous_minutes_overdue=None, instance=None, resources=None)[source]

experimental This API may break in future versions, even between dot releases.

Builds freshness policy sensor context from provided parameters.

This function can be used to provide the context argument when directly invoking a function decorated with @freshness_policy_sensor, such as when writing unit tests.

Parameters:
  • sensor_name (str) – The name of the sensor the context is being constructed for.

  • asset_key (AssetKey) – The AssetKey for the monitored asset

  • freshness_policy (FreshnessPolicy) – The FreshnessPolicy for the monitored asset

  • minutes_overdue (Optional[float]) – How overdue the monitored asset currently is

  • previous_minutes_overdue (Optional[float]) – How overdue the monitored asset was on the previous tick.

  • instance (DagsterInstance) – The dagster instance configured for the context.

Examples

context = build_freshness_policy_sensor_context(
    sensor_name="freshness_policy_sensor_to_invoke",
    asset_key=AssetKey("some_asset"),
    freshness_policy=FreshnessPolicy(maximum_lag_minutes=30)<
    minutes_overdue=10.0,
)
freshness_policy_sensor_to_invoke(context)
@dagster.multi_asset_sensor(monitored_assets, *, job_name=None, name=None, minimum_interval_seconds=None, description=None, job=None, jobs=None, default_status=DefaultSensorStatus.STOPPED, request_assets=None, required_resource_keys=None, tags=None, metadata=None)[source]

experimental This API may break in future versions, even between dot releases.

Creates an asset sensor that can monitor multiple assets.

The decorated function is used as the asset sensor’s evaluation function. The decorated function may:

  1. Return a RunRequest object.

  2. Return a list of RunRequest objects.

  3. Return a SkipReason object, providing a descriptive message of why no runs were requested.

  4. Return nothing (skipping without providing a reason)

  5. Yield a SkipReason or yield one or more RunRequest objects.

Takes a MultiAssetSensorEvaluationContext.

Parameters:
  • monitored_assets (Union[Sequence[AssetKey], AssetSelection]) – The assets this sensor monitors. If an AssetSelection object is provided, it will only apply to assets within the Definitions that this sensor is part of.

  • name (Optional[str]) – The name of the sensor. Defaults to the name of the decorated function.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job to be executed when the sensor fires.

  • jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]]) – (experimental) A list of jobs to be executed when the sensor fires.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • request_assets (Optional[AssetSelection]) – (Experimental) an asset selection to launch a run for if the sensor condition is met. This can be provided instead of specifying a job.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

class dagster.MultiAssetSensorDefinition(name, monitored_assets, job_name, asset_materialization_fn, minimum_interval_seconds=None, description=None, job=None, jobs=None, default_status=DefaultSensorStatus.STOPPED, request_assets=None, required_resource_keys=None, tags=None, metadata=None)[source]

experimental This API may break in future versions, even between dot releases.

Define an asset sensor that initiates a set of runs based on the materialization of a list of assets.

Users should not instantiate this object directly. To construct a MultiAssetSensorDefinition, use dagster. multi_asset_sensor().

Parameters:
  • name (str) – The name of the sensor to create.

  • asset_keys (Sequence[AssetKey]) – The asset_keys this sensor monitors.

  • asset_materialization_fn (Callable[[MultiAssetSensorEvaluationContext], Union[Iterator[Union[RunRequest, SkipReason]], RunRequest, SkipReason]]) –

    The core evaluation function for the sensor, which is run at an interval to determine whether a run should be launched or not. Takes a MultiAssetSensorEvaluationContext.

    This function must return a generator, which must yield either a single SkipReason or one or more RunRequest objects.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job object to target with this sensor.

  • jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]]) – (experimental) A list of jobs to be executed when the sensor fires.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • request_assets (Optional[AssetSelection]) – (Experimental) an asset selection to launch a run for if the sensor condition is met. This can be provided instead of specifying a job.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

class dagster.MultiAssetSensorEvaluationContext(instance_ref, monitored_assets, last_tick_completion_time=None, last_run_key=None, cursor=None, repository_name=None, repository_def=None, instance=None, resource_defs=None, definitions=None, last_sensor_start_time=None, log_key=None, sensor_name=None, code_location_origin=None, last_completion_time=None)[source]

experimental This API may break in future versions, even between dot releases.

The context object available as the argument to the evaluation function of a dagster.MultiAssetSensorDefinition.

Users should not instantiate this object directly. To construct a MultiAssetSensorEvaluationContext for testing purposes, use dagster. build_multi_asset_sensor_context().

The MultiAssetSensorEvaluationContext contains a cursor object that tracks the state of consumed event logs for each monitored asset. For each asset, the cursor stores the storage ID of the latest materialization that has been marked as “consumed” (via a call to advance_cursor) in a latest_consumed_event_id field.

For each monitored asset, the cursor will store the latest unconsumed event ID for up to 25 partitions. Each event ID must be before the latest_consumed_event_id field for the asset.

Events marked as consumed via advance_cursor will be returned in future ticks until they are marked as consumed.

To update the cursor to the latest materialization and clear the unconsumed events, call advance_all_cursors.

monitored_assets

The assets monitored by the sensor. If an AssetSelection object is provided, it will only apply to assets within the Definitions that this sensor is part of.

Type:

Union[Sequence[AssetKey], AssetSelection]

repository_def

The repository that the sensor belongs to. If needed by the sensor top-level resource definitions will be pulled from this repository. You can provide either this or definitions.

Type:

Optional[RepositoryDefinition]

instance_ref

The serialized instance configured to run the schedule

Type:

Optional[InstanceRef]

cursor

The cursor, passed back from the last sensor evaluation via the cursor attribute of SkipReason and RunRequest. Must be a dictionary of asset key strings to a stringified tuple of (latest_event_partition, latest_event_storage_id, trailing_unconsumed_partitioned_event_ids).

Type:

Optional[str]

last_tick_completion_time

The last time that the sensor was evaluated for a tick (UTC).

Type:

Optional[float]

last_run_key

DEPRECATED The run key of the RunRequest most recently created by this sensor. Use the preferred cursor attribute instead.

Type:

str

repository_name

The name of the repository that the sensor belongs to.

Type:

Optional[str]

instance

The deserialized instance can also be passed in directly (primarily useful in testing contexts).

Type:

Optional[DagsterInstance]

definitions

Definitions object that the sensor is defined in. If needed by the sensor, top-level resource definitions will be pulled from these definitions. You can provide either this or repository_def.

Type:

Optional[Definitions]

last_sensor_start_time

The last time the sensor was started.

Type:

Optional[float]

log_key

The log key to use for this sensor tick.

Type:

Optional[List[str]]

sensor_name

The name of the sensor, used for logging and error messages.

Type:

Optional[str]

code_location_origin

The code location that the sensor is in.

Type:

Optional[CodeLocationOrigin]

Example

from dagster import multi_asset_sensor, MultiAssetSensorEvaluationContext

@multi_asset_sensor(monitored_assets=[AssetKey("asset_1), AssetKey("asset_2)])
def the_sensor(context: MultiAssetSensorEvaluationContext):
    ...
advance_all_cursors()[source]

Updates the cursor to the most recent materialization event for all assets monitored by the multi_asset_sensor.

Marks all materialization events as consumed by the sensor, including unconsumed events.

advance_cursor(materialization_records_by_key)[source]

Marks the provided materialization records as having been consumed by the sensor.

At the end of the tick, the cursor will be updated to advance past all materializations records provided via advance_cursor. In the next tick, records that have been consumed will no longer be returned.

Passing a partitioned materialization record into this function will mark prior materializations with the same asset key and partition as having been consumed.

Parameters:

materialization_records_by_key (Mapping[AssetKey, Optional[EventLogRecord]]) – Mapping of AssetKeys to EventLogRecord or None. If an EventLogRecord is provided, the cursor for the AssetKey will be updated and future calls to fetch asset materialization events will not fetch this event again. If None is provided, the cursor for the AssetKey will not be updated.

all_partitions_materialized(asset_key, partitions=None)[source]

A utility method to check if a provided list of partitions have been materialized for a particular asset. This method ignores the cursor and checks all materializations for the asset.

Parameters:
  • asset_key (AssetKey) – The asset to check partitions for.

  • partitions (Optional[Sequence[str]]) – A list of partitions to check. If not provided, all partitions for the asset will be checked.

Returns:

True if all selected partitions have been materialized, False otherwise.

Return type:

bool

get_cursor_partition(asset_key)[source]

A utility method to get the current partition the cursor is on.

get_downstream_partition_keys(partition_key, from_asset_key, to_asset_key)[source]

Converts a partition key from one asset to the corresponding partition key in a downstream asset. Uses the existing partition mapping between the upstream asset and the downstream asset if it exists, otherwise, uses the default partition mapping.

Parameters:
  • partition_key (str) – The partition key to convert.

  • from_asset_key (AssetKey) – The asset key of the upstream asset, which the provided partition key belongs to.

  • to_asset_key (AssetKey) – The asset key of the downstream asset. The provided partition key will be mapped to partitions within this asset.

Returns:

A list of the corresponding downstream partitions in to_asset_key that

partition_key maps to.

Return type:

Sequence[str]

get_trailing_unconsumed_events(asset_key)[source]

Fetches the unconsumed events for a given asset key. Returns only events before the latest consumed event ID for the given asset. To mark an event as consumed, pass the event to advance_cursor. Returns events in ascending order by storage ID.

Parameters:

asset_key (AssetKey) – The asset key to get unconsumed events for.

Returns:

The unconsumed events for the given asset key.

Return type:

Sequence[EventLogRecord]

latest_materialization_records_by_key(asset_keys=None)[source]

Fetches the most recent materialization event record for each asset in asset_keys. Only fetches events after the latest consumed event ID for the given asset key.

Parameters:

asset_keys (Optional[Sequence[AssetKey]]) – list of asset keys to fetch events for. If not specified, the latest materialization will be fetched for all assets the multi_asset_sensor monitors.

Returns: Mapping of AssetKey to EventLogRecord where the EventLogRecord is the latest

materialization event for the asset. If there is no materialization event for the asset, the value in the mapping will be None.

latest_materialization_records_by_partition(asset_key, after_cursor_partition=False)[source]

Given an asset, returns a mapping of partition key to the latest materialization event for that partition. Fetches only materializations that have not been marked as “consumed” via a call to advance_cursor.

Parameters:
  • asset_key (AssetKey) – The asset to fetch events for.

  • after_cursor_partition (Optional[bool]) – If True, only materializations with partitions after the cursor’s current partition will be returned. By default, set to False.

Returns:

Mapping of AssetKey to a mapping of partitions to EventLogRecords where the EventLogRecord is the most recent materialization event for the partition. The mapping preserves the order that the materializations occurred.

Return type:

Mapping[str, EventLogRecord]

Example

@asset(partitions_def=DailyPartitionsDefinition("2022-07-01"))
def july_asset():
    return 1

@multi_asset_sensor(asset_keys=[july_asset.key])
def my_sensor(context):
    context.latest_materialization_records_by_partition(july_asset.key)

# After materializing july_asset for 2022-07-05, latest_materialization_by_partition
# returns {"2022-07-05": EventLogRecord(...)}
latest_materialization_records_by_partition_and_asset()[source]

Finds the most recent unconsumed materialization for each partition for each asset monitored by the sensor. Aggregates all materializations into a mapping of partition key to a mapping of asset key to the materialization event for that partition.

For example, if the sensor monitors two partitioned assets A and B that are materialized for partition_x after the cursor, this function returns:

{
    "partition_x": {asset_a.key: EventLogRecord(...), asset_b.key: EventLogRecord(...)}
}

This method can only be called when all monitored assets are partitioned and share the same partition definition.

materialization_records_for_key(asset_key, limit=None)[source]

Fetches asset materialization event records for asset_key, with the earliest event first.

Only fetches events after the latest consumed event ID for the given asset key.

Parameters:
  • asset_key (AssetKey) – The asset to fetch materialization events for

  • limit (Optional[int]) – The number of events to fetch

property asset_keys

The asset keys which are monitored by this sensor.

Type:

Sequence[AssetKey]

property assets_defs_by_key

A mapping from AssetKey to the AssetsDefinition object which produces it. If a given asset is monitored by this sensor, but is not produced within the same code location as this sensor, then the value will be None.

Type:

Mapping[AssetKey, Optional[AssetsDefinition]]

dagster.build_multi_asset_sensor_context(*, monitored_assets, repository_def=None, instance=None, cursor=None, repository_name=None, cursor_from_latest_materializations=False, resources=None, definitions=None, last_sensor_start_time=None)[source]

experimental This API may break in future versions, even between dot releases.

Builds multi asset sensor execution context for testing purposes using the provided parameters.

This function can be used to provide a context to the invocation of a multi asset sensor definition. If provided, the dagster instance must be persistent; DagsterInstance.ephemeral() will result in an error.

Parameters:
  • monitored_assets (Union[Sequence[AssetKey], AssetSelection]) – The assets monitored by the sensor. If an AssetSelection object is provided, it will only apply to assets within the Definitions that this sensor is part of.

  • repository_def (RepositoryDefinition) – RepositoryDefinition object that the sensor is defined in. Must provide definitions if this is not provided.

  • instance (Optional[DagsterInstance]) – The dagster instance configured to run the sensor.

  • cursor (Optional[str]) – A string cursor to provide to the evaluation of the sensor. Must be a dictionary of asset key strings to ints that has been converted to a json string

  • repository_name (Optional[str]) – The name of the repository that the sensor belongs to.

  • cursor_from_latest_materializations (bool) – If True, the cursor will be set to the latest materialization for each monitored asset. By default, set to False.

  • resources (Optional[Mapping[str, object]]) – The resource definitions to provide to the sensor.

  • definitions (Optional[Definitions]) – Definitions object that the sensor is defined in. Must provide repository_def if this is not provided.

Examples

with instance_for_test() as instance:
    context = build_multi_asset_sensor_context(
        monitored_assets=[AssetKey("asset_1"), AssetKey("asset_2")],
        instance=instance,
    )
    my_asset_sensor(context)
class dagster.RunStatusSensorDefinition(name, run_status, run_status_sensor_fn, minimum_interval_seconds=None, description=None, monitored_jobs=None, monitor_all_code_locations=None, default_status=DefaultSensorStatus.STOPPED, request_job=None, request_jobs=None, tags=None, metadata=None, required_resource_keys=None)[source]

Define a sensor that reacts to a given status of job execution, where the decorated function will be evaluated when a run is at the given status.

Parameters:
  • name (str) – The name of the sensor. Defaults to the name of the decorated function.

  • run_status (DagsterRunStatus) – The status of a run which will be monitored by the sensor.

  • run_status_sensor_fn (Callable[[RunStatusSensorContext], Union[SkipReason, DagsterRunReaction]]) – The core evaluation function for the sensor. Takes a RunStatusSensorContext.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • monitored_jobs (Optional[List[Union[JobDefinition, GraphDefinition, UnresolvedAssetJobDefinition, JobSelector, RepositorySelector, CodeLocationSelector]]]) – The jobs in the current repository that will be monitored by this sensor. Defaults to None, which means the alert will be sent when any job in the repository fails.

  • monitor_all_code_locations (bool) – If set to True, the sensor will monitor all runs in the Dagster deployment. If set to True, an error will be raised if you also specify monitored_jobs or job_selection. Defaults to False.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • request_job (Optional[Union[GraphDefinition, JobDefinition]]) – The job a RunRequest should execute if yielded from the sensor.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

  • request_jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition]]]) – (experimental) A list of jobs to be executed if RunRequests are yielded from the sensor.

class dagster.RunStatusSensorContext(sensor_name, dagster_run, dagster_event, instance, context=None, resource_defs=None, logger=None, partition_key=None, repository_def=None, _resources=None, _cm_scope_entered=False)[source]

The context object available to a decorated function of run_status_sensor.

property dagster_event

The event associated with the job run status.

property dagster_run

The run of the job.

property instance

The current instance.

property log

The logger for the current sensor evaluation.

property partition_key

The partition key of the relevant run.

Type:

Optional[str]

property sensor_name

The name of the sensor.

class dagster.RunFailureSensorContext(sensor_name, dagster_run, dagster_event, instance, context=None, resource_defs=None, logger=None, partition_key=None, repository_def=None, _resources=None, _cm_scope_entered=False)[source]

The context object available to a decorated function of run_failure_sensor.

sensor_name

the name of the sensor.

Type:

str

dagster_run

the failed run.

Type:

DagsterRun

get_step_failure_events()[source]

The step failure event for each step in the run that failed.

Examples

error_strings_by_step_key = {
    # includes the stack trace
    event.step_key: event.event_specific_data.error.to_string()
    for event in context.get_step_failure_events()
}
property failure_event

The run failure event.

If the run failed because of an error inside a step, get_step_failure_events will have more details on the step failure.

class dagster.JobSelector(location_name, repository_name=None, job_name=None)[source]
class dagster.RepositorySelector(*args, **kwargs)[source]
dagster.build_run_status_sensor_context(sensor_name, dagster_event, dagster_instance, dagster_run, context=None, resources=None, partition_key=None, *, repository_def=None)[source]

Builds run status sensor context from provided parameters.

This function can be used to provide the context argument when directly invoking a function decorated with @run_status_sensor or @run_failure_sensor, such as when writing unit tests.

Parameters:
  • sensor_name (str) – The name of the sensor the context is being constructed for.

  • dagster_event (DagsterEvent) – A DagsterEvent with the same event type as the one that triggers the run_status_sensor

  • dagster_instance (DagsterInstance) – The dagster instance configured for the context.

  • dagster_run (DagsterRun) – DagsterRun object from running a job

  • resources (Optional[Mapping[str, object]]) – A dictionary of resources to be made available to the sensor.

  • repository_def (Optional[RepositoryDefinition]) – experimental (This parameter may break in future versions, even between dot releases.) The repository that the sensor belongs to.

Examples

instance = DagsterInstance.ephemeral()
result = my_job.execute_in_process(instance=instance)

dagster_run = result.dagster_run
dagster_event = result.get_job_success_event() # or get_job_failure_event()

context = build_run_status_sensor_context(
    sensor_name="run_status_sensor_to_invoke",
    dagster_instance=instance,
    dagster_run=dagster_run,
    dagster_event=dagster_event,
)
run_status_sensor_to_invoke(context)
@dagster.run_status_sensor(run_status, name=None, minimum_interval_seconds=None, description=None, monitored_jobs=None, job_selection=None, monitor_all_code_locations=None, default_status=DefaultSensorStatus.STOPPED, request_job=None, request_jobs=None, monitor_all_repositories=None, tags=None, metadata=None)[source]

Creates a sensor that reacts to a given status of job execution, where the decorated function will be run when a job is at the given status.

Takes a RunStatusSensorContext.

Parameters:
  • run_status (DagsterRunStatus) – The status of run execution which will be monitored by the sensor.

  • name (Optional[str]) – The name of the sensor. Defaults to the name of the decorated function.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • monitored_jobs (Optional[List[Union[JobDefinition, GraphDefinition, UnresolvedAssetJobDefinition, RepositorySelector, JobSelector, CodeLocationSelector]]]) – Jobs in the current code locations that will be monitored by this sensor. Defaults to None, which means the alert will be sent when any job in the code location matches the requested run_status. Jobs in external repositories can be monitored by using RepositorySelector or JobSelector.

  • monitor_all_code_locations (Optional[bool]) – If set to True, the sensor will monitor all runs in the Dagster deployment. If set to True, an error will be raised if you also specify monitored_jobs or job_selection. Defaults to False.

  • job_selection (Optional[List[Union[JobDefinition, GraphDefinition, RepositorySelector, JobSelector, CodeLocationSelector]]]) – deprecated (This parameter will be removed in version 2.0. Use monitored_jobs instead.) (deprecated in favor of monitored_jobs) Jobs in the current code location that will be monitored by this sensor. Defaults to None, which means the alert will be sent when any job in the code location matches the requested run_status.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • request_job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]) – The job that should be executed if a RunRequest is yielded from the sensor.

  • request_jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJobDefinition]]]) – (experimental) A list of jobs to be executed if RunRequests are yielded from the sensor.

  • monitor_all_repositories (Optional[bool]) – deprecated (This parameter will be removed in version 2.0. Use monitor_all_code_locations instead.) (deprecated in favor of monitor_all_code_locations) If set to True, the sensor will monitor all runs in the Dagster instance. If set to True, an error will be raised if you also specify monitored_jobs or job_selection. Defaults to False.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

@dagster.run_failure_sensor(name=None, minimum_interval_seconds=None, description=None, monitored_jobs=None, job_selection=None, monitor_all_code_locations=None, default_status=DefaultSensorStatus.STOPPED, request_job=None, request_jobs=None, monitor_all_repositories=None, tags=None, metadata=None)[source]

Creates a sensor that reacts to job failure events, where the decorated function will be run when a run fails.

Takes a RunFailureSensorContext.

Parameters:
  • name (Optional[str]) – The name of the job failure sensor. Defaults to the name of the decorated function.

  • minimum_interval_seconds (Optional[int]) – The minimum number of seconds that will elapse between sensor evaluations.

  • description (Optional[str]) – A human-readable description of the sensor.

  • monitored_jobs (Optional[List[Union[JobDefinition, GraphDefinition, UnresolvedAssetJobDefinition, RepositorySelector, JobSelector, CodeLocationSelector]]]) – The jobs in the current repository that will be monitored by this failure sensor. Defaults to None, which means the alert will be sent when any job in the current repository fails.

  • monitor_all_code_locations (bool) – If set to True, the sensor will monitor all runs in the Dagster deployment. If set to True, an error will be raised if you also specify monitored_jobs or job_selection. Defaults to False.

  • job_selection (Optional[List[Union[JobDefinition, GraphDefinition, RepositorySelector, JobSelector, CodeLocationSelector]]]) – deprecated (This parameter will be removed in version 2.0. Use monitored_jobs instead.) (deprecated in favor of monitored_jobs) The jobs in the current repository that will be monitored by this failure sensor. Defaults to None, which means the alert will be sent when any job in the repository fails.

  • default_status (DefaultSensorStatus) – Whether the sensor starts as running or not. The default status can be overridden from the Dagster UI or via the GraphQL API.

  • request_job (Optional[Union[GraphDefinition, JobDefinition, UnresolvedAssetJob]]) – The job a RunRequest should execute if yielded from the sensor.

  • request_jobs (Optional[Sequence[Union[GraphDefinition, JobDefinition, UnresolvedAssetJob]]]) – (experimental) A list of jobs to be executed if RunRequests are yielded from the sensor.

  • monitor_all_repositories (bool) – deprecated (This parameter will be removed in version 2.0. Use monitor_all_code_locations instead.) (deprecated in favor of monitor_all_code_locations) If set to True, the sensor will monitor all runs in the Dagster instance. If set to True, an error will be raised if you also specify monitored_jobs or job_selection. Defaults to False.

  • tags (Optional[Mapping[str, str]]) – A set of key-value tags that annotate the sensor and can be used for searching and filtering in the UI.

  • metadata (Optional[Mapping[str, object]]) – A set of metadata entries that annotate the sensor. Values will be normalized to typed MetadataValue objects.

class dagster.SensorResult(run_requests=None, skip_reason=None, cursor=None, dynamic_partitions_requests=None, asset_events=None, **kwargs)[source]

The result of a sensor evaluation.

run_requests

A list of run requests to be executed.

Type:

Optional[Sequence[RunRequest]]

skip_reason

A skip message indicating why sensor evaluation was skipped.

Type:

Optional[Union[str, SkipReason]]

cursor

The cursor value for this sensor, which will be provided on the context for the next sensor evaluation.

Type:

Optional[str]

dynamic_partitions_requests (Optional[Sequence[Union[DeleteDynamicPartitionsRequest,

AddDynamicPartitionsRequest]]]): A list of dynamic partition requests to request dynamic partition addition and deletion. Run requests will be evaluated using the state of the partitions with these changes applied. We recommend limiting partition additions and deletions to a maximum of 25K partitions per sensor evaluation, as this is the maximum recommended partition limit per asset.

asset_events

(Experimental) A list of materializations, observations, and asset check evaluations that the system will persist on your behalf at the end of sensor evaluation. These events will be not be associated with any particular run, but will be queryable and viewable in the asset catalog.

Type:

Optional[Sequence[Union[AssetObservation, AssetMaterialization, AssetCheckEvaluation]]]

class dagster.AddDynamicPartitionsRequest(partitions_def_name, partition_keys)[source]

A request to add partitions to a dynamic partitions definition, to be evaluated by a sensor or schedule.

class dagster.DeleteDynamicPartitionsRequest(partitions_def_name, partition_keys)[source]

A request to delete partitions to a dynamic partitions definition, to be evaluated by a sensor or schedule.