Add default store for transactions

[desertaxle]

Add a default store and key for transactions opened via the transaction context manager from inside a flow. The transaction context manager will use a copy of the ResultFactory from the flow or task run context with persist_result set to True. That ResultFactory copy will also use the default result storage block if one was not provided to the flow.

The PREFECT_DEFAULT_RESULT_STORAGE_BLOCK has been updated to default to a LocalFilesystem block with the name containing a slugified version of the hostname of the current machine (to avoid conflicts with other users). This block will be created when a transaction is opened with no store provided.

Example

Run a transaction in flow with default record transaction storage:

from prefect import task, flow
from prefect.transactions import transaction, Transaction, TransactionState


@task
def task1():
    return {"x": 1}


@task1.on_commit
def on_commit_task1(txn: Transaction):
    print(f"Task 1 committed with staged value {txn._staged_value}")


@task
def task2(x):
    return {"y": x["x"] + 3}


@task2.on_commit
def on_commit_task2(txn: Transaction):
    print(f"Task 2 committed with staged value {txn._staged_value}")


@flow
def basic_flow():
    with transaction(key="static") as txn:
        if txn.state == TransactionState.COMMITTED:
            return txn.read()
        x = task1()
        y = task2(x)
        txn.stage(y)
        return y


if __name__ == "__main__":
    print(basic_flow())

First run with log the following:

15:11:19.166 | INFO    | prefect.engine - Created flow run 'strict-firefly' for flow 'basic-flow'
15:11:19.167 | WARNING | prefect.utilities.urls - No URL found for the Prefect UI, and no default base path provided.
15:11:19.283 | INFO    | prefect.engine - Created task run 'task1-0' for task 'task1'
15:11:19.334 | INFO    | Task run 'task1-0' - Finished in state Completed()
15:11:19.346 | INFO    | prefect.engine - Created task run 'task2-0' for task 'task2'
15:11:19.393 | INFO    | Task run 'task2-0' - Finished in state Completed()
Task 1 committed with staged value type='unpersisted' artifact_type='result' artifact_description='Unpersisted result of type `dict`'
Task 2 committed with staged value type='unpersisted' artifact_type='result' artifact_description='Unpersisted result of type `dict`'
15:11:19.403 | INFO    | Flow run 'strict-firefly' - Finished in state Completed()
{'y': 4}

Subsequent runs will log the following (previous transaction value will be used):

15:12:38.479 | INFO    | prefect.engine - Created flow run 'porcelain-husky' for flow 'basic-flow'
15:12:38.479 | WARNING | prefect.utilities.urls - No URL found for the Prefect UI, and no default base path provided.
15:12:38.577 | INFO    | Flow run 'porcelain-husky' - Finished in state Completed()
{'y': 4}

Checklist

• This pull request references any related issue by including "closes <link to issue>"
  • 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
• This pull request includes a label categorizing the change e.g. maintenance, fix, feature, enhancement, docs.

For documentation changes:

• This pull request includes redirect settings in mint.json for files that are removed or renamed.

For new functions or classes in the Python SDK:

• This pull request includes helpful docstrings.
• If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in docs/mint.json navigation.

[cicdw]

Nice! one nit about the default name, otherwise LGTM

[cicdw]

does this need an overwrite flag?

[desertaxle]

It shouldn't because we only run this save after confirming it hasn't been loaded from the server, but I'll add overwrite=True to be safe since that should guard against race conditions.