Repository: PrefectHQ/prefect
Pull request: Add default store for transactions
<!-- 
Thanks for opening a pull request to Prefect! We've got a few requests to help us review contributions:

- Make sure that your title neatly summarizes the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share an example to help us understand the change in user experience.
- Confirm that you've done common tasks so we can give a timely review.
- Review our contribution guidelines: https://docs.prefect.io/latest/contributing/overview/

Happy engineering!
-->

<!-- Include an overview here -->
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
<!-- 
Share an example of the change in action.

A code blurb is best. Changes to features should include an example that is executable by a new user.
If changing documentation, a link to a preview of the page is great.
 -->
Run a transaction in flow with default record transaction storage:
```python
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:
```python
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):
```python
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
<!-- These boxes may be checked after opening the pull request. -->

- [ ] 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](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] If this pull request adds new functionality, it includes unit tests that cover the changes
- [x] This pull request includes a label categorizing the change e.g. `maintenance`, `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

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.


Problem statement:
Add default local storage persistence for transactions
Setting up a transaction store is difficult because it requires creating a result store, which requires creating a block. Without a store, transaction results are not persisted. We should be able to set up a default transaction store for users that uses local storage. We should not create an anonymous block for this case, but instead create and reuse a default transaction store block. Transaction results should be isolated by default to ensure no result reuse between runs unless the user specifically enables it via their transaction key.

Hints from issue discussion:
Just a thought: we could allow configuring the directory where results are stored along with its structure via a setting for easy testing and local dev.
I like that idea! Can you elaborate on configuring the structure via a setting? What are some examples of structures that we might offer?
I might be a one trick pony on templating, but I was imagining things like:
```
PREFECT_LOCAL_STORAGE_PATH="~/.prefect/{flow_run_id}/"
```
as an example.

Commit message: Add default store and key for `transaction` context manager

Patch to review:
diff --git a/src/prefect/context.py b/src/prefect/context.py
index 544ef2d96b5e..4f04a6d2bab1 100644
--- a/src/prefect/context.py
+++ b/src/prefect/context.py
@@ -151,7 +151,7 @@ def __enter__(self):
         self._token = self.__var__.set(self)
         return self
 
-    def __exit__(self, *_):
+    def __exit__(self, exc_type, exc_val, exc_tb):
         if not self._token:
             raise RuntimeError(
                 "Asymmetric use of context. Context exit called without an enter."
diff --git a/src/prefect/results.py b/src/prefect/results.py
index e69a864fd4f6..6aaa42a07275 100644
--- a/src/prefect/results.py
+++ b/src/prefect/results.py
@@ -62,15 +62,17 @@ def DEFAULT_STORAGE_KEY_FN():
 
 
 @sync_compatible
-async def get_default_result_storage() -> ResultStorage:
+async def get_default_result_storage() -> WritableFileSystem:
     """
     Generate a default file system for result storage.
     """
-    return (
-        await Block.load(PREFECT_DEFAULT_RESULT_STORAGE_BLOCK.value())
-        if PREFECT_DEFAULT_RESULT_STORAGE_BLOCK.value() is not None
-        else LocalFileSystem(basepath=PREFECT_LOCAL_STORAGE_PATH.value())
-    )
+    try:
+        return await Block.load(PREFECT_DEFAULT_RESULT_STORAGE_BLOCK.value())
+    except ValueError:
+        block = LocalFileSystem(basepath=PREFECT_LOCAL_STORAGE_PATH.value())
+        default_name = PREFECT_DEFAULT_RESULT_STORAGE_BLOCK.value().split("/")[-1]
+        await block.save(default_name)
+        return block
 
 
 _default_task_scheduling_storages: Dict[Tuple[str, str], WritableFileSystem] = {}
diff --git a/src/prefect/settings.py b/src/prefect/settings.py
index accf43c588ec..1f148f6c07ec 100644
--- a/src/prefect/settings.py
+++ b/src/prefect/settings.py
@@ -42,6 +42,7 @@
 
 import logging
 import os
+import socket
 import string
 import warnings
 from contextlib import contextmanager
@@ -84,6 +85,7 @@
 from prefect.exceptions import MissingProfileError
 from prefect.utilities.names import OBFUSCATED_PREFIX, obfuscate
 from prefect.utilities.pydantic import add_cloudpickle_reduction
+from prefect.utilities.slugify import slugify
 
 T = TypeVar("T")
 
@@ -417,6 +419,18 @@ def warn_on_misconfigured_api_url(values):
     return values
 
 
+def default_result_storage_block_name(settings, value):
+    """
+    `value_callback` for `PREFECT_DEFAULT_RESULT_STORAGE_BLOCK_NAME` that sets the default
+    value to the hostname of the machine.
+    """
+    if value is None:
+        return (
+            f"local-file-system/{slugify(socket.gethostname())}-default-local-storage"
+        )
+    return value
+
+
 def default_database_connection_url(settings, value):
     templater = template_with_settings(PREFECT_HOME, PREFECT_API_DATABASE_PASSWORD)
 
@@ -1575,8 +1589,7 @@ def default_cloud_ui_url(settings, value):
 # Defaults -----------------------------------------------------------------------------
 
 PREFECT_DEFAULT_RESULT_STORAGE_BLOCK = Setting(
-    Optional[str],
-    default=None,
+    Optional[str], default=None, value_callback=default_result_storage_block_name
 )
 """The `block-type/block-document` slug of a block to use as the default result storage."""
 
diff --git a/src/prefect/transactions.py b/src/prefect/transactions.py
index 2826a10134db..b73feb66e9b0 100644
--- a/src/prefect/transactions.py
+++ b/src/prefect/transactions.py
@@ -1,3 +1,4 @@
+import uuid
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
 from typing import (
@@ -7,17 +8,17 @@
     List,
     Optional,
     Type,
-    TypeVar,
 )
 
 from pydantic import Field
+from typing_extensions import Self
 
-from prefect.context import ContextModel
+from prefect.context import ContextModel, FlowRunContext
 from prefect.records import RecordStore
+from prefect.records.result_store import ResultFactoryStore
+from prefect.results import get_default_result_storage
 from prefect.utilities.collections import AutoEnum
 
-T = TypeVar("T")
-
 
 class IsolationLevel(AutoEnum):
     READ_COMMITTED = AutoEnum.auto()
@@ -54,7 +55,7 @@ class Transaction(ContextModel):
     )
     overwrite: bool = False
     _staged_value: Any = None
-    __var__ = ContextVar("transaction")
+    __var__: ContextVar = ContextVar("transaction")
 
     def is_committed(self) -> bool:
         return self.state == TransactionState.COMMITTED
@@ -123,11 +124,19 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def begin(self):
         # currently we only support READ_COMMITTED isolation
         # i.e., no locking behavior
-        if not self.overwrite and self.store and self.store.exists(key=self.key):
+        if (
+            not self.overwrite
+            and self.store
+            and self.key
+            and self.store.exists(key=self.key)
+        ):
             self.state = TransactionState.COMMITTED
 
     def read(self) -> dict:
-        return self.store.read(key=self.key)
+        if self.store and self.key:
+            return self.store.read(key=self.key)
+        else:
+            return {}
 
     def reset(self) -> None:
         parent = self.get_parent()
@@ -136,8 +145,9 @@ def reset(self) -> None:
             # parent takes responsibility
             parent.add_child(self)
 
-        self.__var__.reset(self._token)
-        self._token = None
+        if self._token:
+            self.__var__.reset(self._token)
+            self._token = None
 
         # do this below reset so that get_transaction() returns the relevant txn
         if parent and self.state == TransactionState.ROLLED_BACK:
@@ -165,7 +175,7 @@ def commit(self) -> bool:
             for hook in self.on_commit_hooks:
                 hook(self)
 
-            if self.store:
+            if self.store and self.key:
                 self.store.write(key=self.key, value=self._staged_value)
             self.state = TransactionState.COMMITTED
             return True
@@ -174,11 +184,17 @@ def commit(self) -> bool:
             return False
 
     def stage(
-        self, value: dict, on_rollback_hooks: list, on_commit_hooks: list
+        self,
+        value: dict,
+        on_rollback_hooks: Optional[List] = None,
+        on_commit_hooks: Optional[List] = None,
     ) -> None:
         """
         Stage a value to be committed later.
         """
+        on_commit_hooks = on_commit_hooks or []
+        on_rollback_hooks = on_rollback_hooks or []
+
         if self.state != TransactionState.COMMITTED:
             self._staged_value = value
             self.on_rollback_hooks += on_rollback_hooks
@@ -203,11 +219,11 @@ def rollback(self) -> bool:
             return False
 
     @classmethod
-    def get_active(cls: Type[T]) -> Optional[T]:
+    def get_active(cls: Type[Self]) -> Optional[Self]:
         return cls.__var__.get(None)
 
 
-def get_transaction() -> Transaction:
+def get_transaction() -> Optional[Transaction]:
     return Transaction.get_active()
 
 
@@ -218,6 +234,35 @@ def transaction(
     commit_mode: CommitMode = CommitMode.LAZY,
     overwrite: bool = False,
 ) -> Generator[Transaction, None, None]:
+    """
+    A context manager for opening and managing a transaction.
+
+    Args:
+        - key: An identifier to use for the transaction
+        - store: The store to use for persisting the transaction result
+        - commit_mode: The commit mode controlling when the transaction and
+            child transactions are committed
+        - overwrite: Whether to overwrite an existing transaction record in the store
+
+    Yields:
+        - Transaction: An object representing the transaction state
+    """
+    if not store and (flow_run_context := FlowRunContext.get()):
+        if flow_run_context:
+            new_factory = flow_run_context.result_factory.model_copy(
+                update={
+                    "persist_result": True,
+                }
+            )
+            if not new_factory.storage_block:
+                default_storage = get_default_result_storage(_sync=True)
+                new_factory.storage_block = default_storage
+            store = ResultFactoryStore(
+                result_factory=new_factory,
+            )
+
+    if not key:
+        key = str(uuid.uuid4())
     with Transaction(
         key=key, store=store, commit_mode=commit_mode, overwrite=overwrite
     ) as txn:


Code review comments:
diff_hunk: @@ -218,6 +234,35 @@ def transaction(
     commit_mode: CommitMode = CommitMode.LAZY,
     overwrite: bool = False,
 ) -> Generator[Transaction, None, None]:
+    """
+    A context manager for opening and managing a transaction.
+
+    Args:
+        - key: An identifier to use for the transaction
+        - store: The store to use for persisting the transaction result
+        - commit_mode: The commit mode controlling when the transaction and
+            child transactions are committed
+        - overwrite: Whether to overwrite an existing transaction record in the store
+
+    Yields:
+        - Transaction: An object representing the transaction state
+    """
+    if not store and (flow_run_context := FlowRunContext.get()):
+        if flow_run_context:
+            new_factory = flow_run_context.result_factory.model_copy(
+                update={
+                    "persist_result": True,
+                }
+            )
+            if not new_factory.storage_block:
+                default_storage = get_default_result_storage(_sync=True)
+                new_factory.storage_block = default_storage
+            store = ResultFactoryStore(
+                result_factory=new_factory,
+            )
+
+    if not key:
+        key = str(uuid.uuid4())
path: src/prefect/transactions.py
text: @user1:
This unfortunately won't work in certain situations - the one that immediately comes to mind is if a task has a `NONE` cache policy; is there a reason you needed this default?  The task engine should always be providing a relevant key

@author:
I added this default so that using the `transaction` context manager directly would persist a record by default, but I realize now that doesn't make sense. I'll remove this.