Support direct asset materialization through Python function

[Jay-Lokhande]

Checklist

• This pull request references any related issue by including "closes #19634 "

  • If no issue exists and your change is not a small fix, please create an issue first.

• If this pull request adds new functionality, it includes unit tests that cover the changes

• If this pull request removes docs files, it includes redirect settings in mint.json.

• If this pull request adds functions or classes, it includes helpful docstrings.

• Modified materialize() function to support both decorator and direct call usage

• When called directly in execution context (flow or task), assets are materialized immediately

• Task context is automatically inherited when materializing from within a task

• Assets materialized directly become proper upstream dependencies for downstream tasks/assets

closes #19634

[codspeed-hq]

CodSpeed Performance Report

Merging this PR will not alter performance

_{Comparing Jay-Lokhande:feat/direct-asset-materialization-19634 (492e325) with main (964b4ca)}

Summary

✅ 2 untouched benchmarks

[robfreedy]

@Jay-Lokhande are you still working on this PR? We have had some interest in this feature so wanted to follow up to see if this is ready for review or if there is any way we could help!

[Jay-Lokhande]

Hi @robfreedy, thanks for checking in

I am still working on this PR the core implementation for direct asset materialization is complete the materialize() function now supports being called directly from within flows and tasks not just as a decorator
however I am running into an issue with the test suite the tests for flow-level direct materialization are failing because asset materialization events are not being captured by the asserting_events_worker fixture, events appear to be emitted correctly (the code path executes) but they are not showing up in the test assertions

The failing tests are:

test_direct_materialization_from_flow
test_direct_materialization_with_dependencies
test_direct_materialization_with_string_key

all of these tests call materialize() directly within a flow and then check for materialization events but len(mat_events) == 0 instead of the expected count

would you be able to help debug this?
specifically:
Are there any known issues with the asserting_events_worker fixture and flow-level event emission?
Is there a pattern in other tests that successfully emit events from within flows that I should follow?

[zzstoatzz]

hi @Jay-Lokhande! I took a look at the failing tests and think i found the issue

which is that AssetContext (which inherits from ContextModel) doesn't have a set() method - it uses the context manager protocol (__enter__/__exit__) instead. The calls to asset_ctx.set() were raising AttributeError, which was silently caught by the try/except Exception: block in materialize().

i've opened a PR with the fix into your branch: Jay-Lokhande#1

the fix simplifies things a bit:

• Task context: The task engine already sets up an AssetContext via its context manager, so we just get the existing context and add our assets to it
• Flow context: Create a fresh AssetContext per materialize() call and emit events immediately

let me know if that makes sense. feel free to merge it to update your PR here!

[Jay-Lokhande]

@zzstoatzz hi, thank you so much for identifying and fixing the issue
you are right i was incorrectly trying to use asset_ctx.set(), which does not exist on ContextModel subclasses. the fix makes sense

[Jay-Lokhande]

@zzstoatzz would you please give your review on the suggested changes, thanks!

[Jay-Lokhande]

@zzstoatzz Hey!
would you please review the PR

[zzstoatzz]

sorry for the delay @Jay-Lokhande - we'll get this reviewed early next week

[Jay-Lokhande]

@zzstoatzz sure, No worries, and thank you for the reply!

[zzstoatzz]

thanks for bearing with us @Jay-Lokhande, looking mostly good, just a couple things

[zzstoatzz]

this seems like an unreachable case

[zzstoatzz]

i think this is a known flake, that unless was causing all of your CI to fail (and not just one flake), i think we should revert for now

[Jay-Lokhande]

@zzstoatzz Thanks for the feedback! i have addressed both comments.

[zzstoatzz]

this lgtm, cc @desertaxle for another set of eyes

[desertaxle]

I have some concerns with this implementation, one of which I called out on the updated typing of materialize.

Fundamentally, I don't think a function makes sense for direct asset materialization because the operation is so contextual. I think it'd make more sense to have a context manager that lets the materialization logic apply to a code block when an author doesn't want to write an entire function for materialization.

@Jay-Lokhande how does the idea of introducing a context manager instead of a function for direct materialization sit with you?

[desertaxle]

I'm not a fan of how this return type is overloaded as a result of this work, as it'll degrade the typing experience of the @materialize decorator.

I think there should be a separate utility for direct materialization to keep the API clear and explicit here.

[Jay-Lokhande]

Thanks for the feedback! @desertaxle, you are right on both counts - the context manager makes more sense for this, and keeping materialize() as a pure decorator will preserve the typing experience.
I amm happy to refactor this a few questions to make sure I get it right:

• Should the context manager materialize on entry or exit? I am guessing exit makes more sense (after the block completes), but wanted to confirm.

• What should we call the new utility? materialize_assets() or something else? And should materialize() stay as just the decorator?

• Should parameters like by work the same way?

  with materialize_assets("s3://bucket/data.csv", by="dbt"):
      ...

Let me know what you prefer and I will update the PR!

[desertaxle]

@Jay-Lokhande, we've discussed the approach in this PR and the intended Prefect asset materialization design, and we've decided not to accept it.

Right now, asset materialization is intentionally tied to the task lifecycle. Certain aspects of asset materialization (such as dependencies) are difficult to reason about outside a task lifecycle, and we need further discussion to identify an approach that aligns well with the intent of asset materialization.

I'm going to close this PR, but let's continue discussing approaches on #19634 to see if we can find an approach that aligns with the requester's desires and the feature's intent.

[jbnitorum]

@desertaxle Sorry just to chime in here as it relates to my original issue. is there an avenue you are taking in order to materialize arbitrary assets? For example we are seeking a way to generate downstream assets for a materialized assets where the downstream asset doesn't actually have its own task.

Example: We can currently materialize a snowflake table when we update it. How do we materialize 3 Tableau dashboards that utilize that table. We'd like to have a task that takes inputs from a parent @materialize decorated task and then queries our list of related assets and materializes that 3 dashboards within that child task, ideally inheriting the upstream asset as intended by the current asset implementation.

If we lock to one task, one asset it makes it impossible to have patterns like this.

Also its beneficial in retrofitting some legacy flows so you don't have to go back and change the decorator.

[desertaxle]

@jbnitorum great question! I'm going to copy your question to the issue and respond there so this discussion doesn't get lost.