AIP-103: Add Task SDK support for retrieving Asset State by `name`/`uri`

[jroachgolf84]

Description

Support AssetState mechanism in the Task SDK for use in things like BaseEventTrigger's, which don't have access to context.

from airflow.sdk import AssetState

asset_state = AssetState(name="my_asset")  # Pass in the name of the Asset (or uri)

asset_state.set("watermark": "1")         # Store a value
watermark = asset_state.get("watermark")  # Retrieve the value

print(watermark)  # Should output "1"

Use case/motivation

Allow for Asset State to be retrieved/set in BaseEventTrigger, useful for Asset-watching.

Related issues

PR: #66595 (comment)

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Codingaditya17]

Hi, I saw this issue is assigned to you. I’m interested in helping with this if you are okay with it.

I can start by investigating where the existing AssetState accessors are implemented and propose adding GetAssetStateByName / GetAssetStateByUri support in the Task SDK, with tests.

Please let me know if you are already actively working on it or if there is a specific part I can take.

[jroachgolf84]

@Codingaditya17, I'll let you know when my PR is ready for review. It's going to be drafted momentarily. I moved away from GetAssetStateByName, the interface is a bit different.

[Codingaditya17]

Sounds like a plan. Thank you

[jroachgolf84]

@Codingaditya17 - here is the first draft of my PR. Feel free to take a look. Let me know your thoughts. Just note, this is a draft and is going to be updated.

#67248

[Codingaditya17]

Thanks for sharing the draft. I did a quick first pass and the overall direction looks good to me. The AssetState(name=...) / AssetState(uri=...) API feels cleaner than exposing separate name/uri getters directly.

Since this is still a draft and CI is currently failing on static checks, docs spellcheck, and Task SDK tests, I would wait before doing a full review. One thing I noticed from the diff is that the ToTriggerRunner union around MaskSecret may need cleanup, since it looks like there may be duplicate MaskSecret / comma placement issues.

Also, since the PR description mentions usage from both Trigger and Task, I think it would be good to add or confirm task-side coverage too, along with validation tests for both/neither name and uri.

Happy to re-review after the next update.

[jroachgolf84]

@Codingaditya17 - thanks for the feedback! Can you copy-and-paste this into the PR itself? Adding the review there also helps to get a little green on your GitHub profile :)