Support Asset materialization through a python function directly #19634
Description
Describe the current behavior
Currently, you can work around materializing an Asset in code without the decorator something like the following code:
from prefect import task, flow
from prefect.assets import Asset, materialize
def get_asset_uris_from_db():
return ["s3://test-assets/test_asset.txt", "s3://test-assets/test_asset2.txt"]
def task_a():
pass
@materialize("s3://test-assets/test_asset3.txt")
def task_b():
pass
@flow
def my_flow():
asset_uris = get_asset_uris_from_db()
for asset_uri in asset_uris:
basic_asset = Asset(key=asset_uri)
materialize(basic_asset)(task_a)()
task_b()
if __name__ == "__main__":
my_flow()Describe the proposed behavior
Allow the materialization of an asset anywhere in the flow code by calling materialize with an Asset and configuration passed in as an argument without having to attach it to a task. Something like the below code could work.
Example Use
@flow
def my_flow():
basic_asset = Asset(key=asset_uri)
materialize(basic_asset)Additional context
No response