What is the recommended pattern for detecting a ma...
# ask-community
l
What is the recommended pattern for detecting a materialization failure associated to an asset that is being materialized by an asset reconciliation sensor? There is no "job" that can be monitored using a job failure sensor, and an asset sensor only listens for successful materializations
🤖 1
Use case is to send a notification when the materialization fails for a subset of assets.
Oh, i see that this may not be possible - https://github.com/dagster-io/dagster/discussions/13958 Is there any downside to having multiple asset reconciliation sensors? It seems like the best way for now would be to configure an asset reconciliation sensor with specific tags and use the Dagster Cloud notification tooling
o
hi @Leo Qin! a few dimensions to this, but the first is that the regular
run_failure_sensor
should work for your purposes, as it (by default) monitors for failures of any of your jobs. You can check which asset keys a job was intended to materialize by looking at the
context.dagster_run.asset_selection
. You can then compare this to the set of assets that the run did materialize via
Copy code
context.instance.get_records_for_run(run_id=run_id, of_type=DagsterEventType.ASSET_MATERIALIZATION).records
The second dimension is that we recommend you use the new Auto Materialize Daemon, rather than
build_asset_reconciliation_sensor
, especially with Dagster Cloud (it works across repositories and is more performant). You can detect runs that were launched by the daemon by looking for the
"dagster/auto_materialize": "true"
tag on the run.
🌈 1
l
ah, thanks! Yeah, we'll be moving towards the Daemon for sure.... Is there a way to access the Definition Metadata for an asset from an AssetKey? The idea here is that we could configure destinations/interested parties for a notification in the definition metadata for an asset.
sorry, @owen - is it possible to access definition metadata from the sensor context?
o
hi @Leo Qin! yep this is possible -- you can always access your
Definitions
(or repository) object by importing it in your run status sensor code, i.e.
Copy code
@run_status_sensor(...)
def my_sensor(...):
    from my_stuff import defs
    asset_graph = defs.get_repository_def().asset_graph
    all_asset_defs = asset_graph.assets # list of all your AssetsDefinitions
from there, you can search that list for the key you're interested in