Hi - Rookie question: I am needing to access all of my asset values from within a single op. How to I get access to the repository (which is in a different top-level file) to call

get_asset_value_loader

without encountering the circular dependencies error?

hi @Adam Ward! sounds like an interesting use case, curious to hear more, but as for your original question the panacea for all circular dependency errors is importing the required element within the body of the function that uses it, so:

@op
def foo():
    from ..repository import my_repo
    with my_repo.get_asset_value_loader() as loader:
        ...

Ahh! Fantastic! Thank you! As for our use case, I'm trying to create a combination of a Freshness Policy Sensor, Failure Sensor, and a scheduled email notification in one. Basically, I want an email notification sent every 15 mins with all assets that are either stale and/or the previous materialization run failed. I also want the email to include the error_info from the RUN_FAILURE event, if applicable. The freshness_policy_sensor is almost the right tool here as I could just add the logic for looking up the RUN_FAILURE event details. Unfortunately, the decorated function runs for each asset in the given AssetSelection, which - in our case - is generating n number of emails. I need the function to run against all assets and send the aggregated info in a single job. My assumption/hope is that I can use the

get_asset_value_loader

function to get the latest run_id and determine whether the given asset is reconciled or not. 🤞🏻

ah interesting -- I don't think

get_asset_value_loader

is the right tool for the job here actually. It just loads the actual value of the asset as an in-memory object. We are planning at some point to add a

batch_freshness_policy_sensor

which would run for all selected assets at once, but that work isn't planned for the immediate future

Rats - Thanks for saving me the work here. 🙂 Would it be possible to define a freshness policy sensor within a scheduled job such that it appends each stale asset to a list or dict in the scheduled job where it can then batch them into an email?

I am loathe to suggest a non-public API that's currently being iterated upon, but you might get some inspiration from the CachingInstanceQueryer. This has a bunch of methods for calculating status of assets, and is what we use internally inside the freshness_policy_sensor

I guess I'm kind of looking for an ephemeral freshness policy sensor within the scope of the scheduled function

So I think your best bet in the current state of the world would be to create your own regular

@sensor

, then use the instance queryer to calculate the status of each of your assets (and use that to create some config dictionary that you add to your RunRequest for the email job)

That's understandable. Thank you very much! Is there an issue or PR somewhere that I can follow for the batch sensor, by chance?