Support Asset materialization through a python function directly

[robfreedy]

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

[jbnitorum]

Thanks for posting Rob. I'd further elaborate that it would be help to call from task context as well (and have the task id/name from within which it is called included in asset metadata). We have a desire to 'materialize' downstream assets dynamically of assets that are materializing in a more traditional sense. A good example is a Tableau workbook. You might be materializing a snowflake table in the most literal sense within a task as the task literally modifies the table, however, something like a Tableau workbook that uses the table as source will change but is not explicitly touched. We want to present ALL assets related to a flow even if a flow doesn't explicitly touch it. Ideally if we can integrate a function like this into an existing function we use to write to snowflake we could

update snowflake -> query for downstream resources -> 'materialize' the downstream assets

with very minimal if any changes to our hundreds of existing flows. While the flow might not touch the workbook Prefect assets does have the potential to be a single pane of glass for all data assets so long as we have an avenue 'materialize' arbitrary assets within flows/tasks.

[jbnitorum]

Further following up on the above in our testing it looks like if you call to materialize an asset from within a task like in the initial example that creates a task within a task and in doing so seems to break the ability for that asset to be automatically an upstream dependency of any other tasks/assets that are downstream of the parent task.

In an ideal world for us you have a function that can materialize an asset within a task and it inherits the task context from the task within which it is being called... similarly any of those assets become upstream dependencies of any assets materialized in any tasks downstream of the first task.

The current implementation seems limit the tasks materialized assets as a 1:1 relationship... only allowing for the task itself to trigger materialization.