In an asset sensor, how can I get the metadata of materialized asset that was detected

In an asset sensor, you can access the metadata of a materialized asset by querying the event log for the latest materialization event of the asset. The documentation provides an example of how to access metadata from a historical materialization of one asset from inside another asset or an op, and a similar approach can be used within an asset sensor. Here's a step-by-step guide based on the documentation: 1. Obtain the

instance

attribute from the sensor's context. This instance is a

DagsterInstance

object that can interact with the event log. 2. Use the

instance

to query for the latest materialization event of the asset using the

get_latest_materialization_event

method, passing in the

AssetKey

of the asset you're interested in. 3. Access the

asset_materialization

property of the event to get the materialization object. 4. Retrieve the metadata from the materialization object's

metadata

property. Here is a conceptual example based on the documentation:

python from dagster import sensor, SensorEvaluationContext, DagsterInstance, AssetKey  @sensor def my_asset_sensor(context: SensorEvaluationContext):     instance = context.instance  # Get the DagsterInstance from the context     asset_key = AssetKey(["my_asset"])  # Replace with your asset's key     materialization_event = instance.get_latest_materialization_event(asset_key)     if materialization_event:         materialization = materialization_event.asset_materialization         metadata = materialization.metadata  # Access the metadata dictionary         # Now you can use the metadata as needed

Please note that the actual code you write may vary depending on the specific details of your use case and the version of Dagster you are using. The example above is a conceptual guide and may require adjustments to fit your context.

@Scout can I get the partition key of the asset

Yes, you can get the partition key of a materialized asset from the materialization event. When you query the event log for the latest materialization event of an asset, the event object contains information about the partition key if the asset is partitioned. Here's how you can access the partition key based on the documentation: 1. Obtain the

instance

attribute from the sensor's context, which is a

DagsterInstance

object. 2. Use the

instance

to query for the latest materialization event of the asset using the

get_latest_materialization_event

method, passing in the

AssetKey

of the asset you're interested in. 3. Access the

event_log_entry

property of the event to get the event log entry object. 4. Retrieve the partition key from the

partition

property of the event log entry object. Here is a conceptual example:

python from dagster import sensor, SensorEvaluationContext, DagsterInstance, AssetKey  @sensor def my_asset_sensor(context: SensorEvaluationContext):     instance = context.instance  # Get the DagsterInstance from the context     asset_key = AssetKey(["my_asset"])  # Replace with your asset's key     materialization_event = instance.get_latest_materialization_event(asset_key)     if materialization_event:         partition_key = materialization_event.event_log_entry.partition  # Access the partition key         # Now you can use the partition_key as needed

This example assumes that the asset is partitioned and that a materialization event for the asset exists. If the asset is not partitioned or if there is no materialization event, the

partition

property may be

None

or the event may not exist. Adjust the code to handle these cases as needed for your application.

@Scout what are the available parameters in a asset_sensor

In Dagster, when you define an

asset_sensor

, you can use several parameters to configure its behavior. The available parameters for an

asset_sensor

decorator are as follows: -

asset_key

(required): The

AssetKey

that the sensor monitors for materialization events. -

name

(optional): The name of the sensor. If not provided, it defaults to the name of the decorated function. -

job_name

(optional): The name of the job to be triggered by the sensor. This parameter is deprecated in favor of the

job

parameter. -

job

(optional): The job that the sensor will trigger when it fires. This replaces the

job_name

parameter. -

jobs

(optional): A list of jobs that the sensor will trigger when it fires. This is useful if you want the sensor to be able to trigger multiple jobs. -

minimum_interval_seconds

(optional): The minimum number of seconds that will elapse between sensor evaluations. This helps prevent sensors from running too frequently. -

description

(optional): A human-readable description of the sensor. -

default_status

(optional): The default status of the sensor (

RUNNING

or

STOPPED

). It determines whether the sensor starts as active or inactive. -

required_resource_keys

(optional): A set of resource keys that the sensor requires to run. This ensures that the necessary resources are available when the sensor executes. Here's an example of defining an

asset_sensor

with some of these parameters:

python from dagster import asset_sensor, AssetKey, RunRequest, SensorEvaluationContext  @asset_sensor(     asset_key=AssetKey("my_asset"),     name="my_asset_sensor",     minimum_interval_seconds=30,     description="Triggers when my_asset is materialized." ) def my_asset_sensor(context: SensorEvaluationContext, asset_event):     # Sensor logic to create a RunRequest or SkipReason     return RunRequest(run_key="unique_run_key", run_config={})

Please note that the parameters and their usage might evolve over time, so it's always a good idea to consult the latest Dagster documentation for the most up-to-date information.