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 (4d23a91) with main (aa753c8)}

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