We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Behaviour differs from documentation: state.result & flow.get_tasks 
Prefect Version: 0.10.4

## Current behavior
Minor issue - The following behaviour doesn't match up to that described by the documentation on [result handling](https://docs.prefect.io/core/concepts/results.html#results):

```python
>>> task_ref = flow.get_tasks[0]
>>> state = flow.run()
>>> state.result[task_ref]._result  # a Task State's Result contains the Task's return value
<Result: 1>
```

Instead prefect requires that the task_ref be created via a name or tag call (`flow.get_tasks(name='<name>')` instead of `flow.get_tasks[0]`) and be extracted from its list (`state.result[task_ref[0]]` instead of state.result[task_ref]):

```python
>>> task_ref = flow.get_tasks(name='<name>')
>>> state = flow.run()
>>> state.result[task_ref[0]]._result  # a Task State's Result contains the Task's return value
<Result: 1>
```

and also the documentation on [Local Debugging](https://docs.prefect.io/core/advanced_tutorials/local-debugging.html#local-debugging):

```python
from prefect import Flow, task

@task
def gotcha():
    tup = ('a', ['b'])
    try:
        tup[1] += ['c']
    except TypeError:
        assert len(tup[1]) == 1

flow = Flow(name="tuples", tasks=[gotcha])

state = flow.run()
state.result # {<Task: gotcha>: Failed("Unexpected error: AssertionError()")}

failed_state = state.result[gotcha]
raise failed_state.result
```

## Examples

### `flow.get_tasks`:

```python
from prefect import Flow, task
import pandas as pd

@task(name='load data')
def _load_data() -> pd.DataFrame:
    return pd.DataFrame({'a': [1, 2, 3]})


with Flow('preprocess-VO') as flow:
    _load_data()

state = flow.run()
task_ref = flow.get_tasks[0]
```

Returns:
```python
>>> TypeError: 'method' object is not subscriptable
```

### `state.result[task_ref]`

```python
from prefect import Flow, task
import pandas as pd

@task(name='load data')
def _load_data() -> pd.DataFrame:
    return pd.DataFrame({'a': [1, 2, 3]})


with Flow('preprocess-VO') as flow:
    _load_data()

state = flow.run()
task_ref = flow.get_tasks(name='load data')
state.result[task_ref]
```

Returns:
```python
>>> TypeError: unhashable type: 'list'
```

### `state.result[<function name>]`

```python
from prefect import Flow, task
import pandas as pd

@task(name='load data')
def _load_data() -> pd.DataFrame:
    return pd.DataFrame({'a': [1, 2, 3]})


with Flow('preprocess-VO') as flow:
    _load_data()

state = flow.run()
task_ref = flow.get_tasks(name='load data')
state.result[_load_data]
```

```python
KeyError: <Task: load data>
```

## Proposed behavior
It may be a bit more readable to adapt the functionality here so that Prefect behaves as in the documentation so:
- `state.result[task_ref]` can be used instead of `state.result[task_ref[0]]` 
- `state.result[<func name>]` works
- `flow.get_tasks[0]` returns a task reference

Additionally, it would be great if function names could be passed to either `state.result[<func name>]` (EDIT: I see now that this was functionality did exist) or `flow.get_tasks(<func name>)` as an alternative to adding tags and/or names to each task function to avoid having to double name all functions.

---

Thanks for creating and maintaining prefect, really enjoying the library so far 


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/engine/results/azure_result.py`
Content:
```
1 import os
2 from typing import TYPE_CHECKING, Any
3 
4 from prefect.client import Secret
5 from prefect.engine.result import Result
6 
7 if TYPE_CHECKING:
8     import azure.storage.blob
9 
10 
11 class AzureResult(Result):
12     """
13     Result for writing to and reading from an Azure Blob storage.
14 
15     Note that your flow's runtime environment must be able to authenticate with
16     Azure; there are currently two supported options: provide a connection string
17     either at initialization or at runtime through an environment variable, or
18     set your Azure connection string as a Prefect Secret.  Using an environment
19     variable is the recommended approach.
20 
21     Args:
22         - container (str): the name of the container to write to / read from
23         - connection_string (str, optional): an Azure connection string for communicating with
24             Blob storage. If not provided the value set in the environment as `AZURE_STORAGE_CONNECTION_STRING`
25             will be used
26         - connection_string_secret (str, optional): the name of a Prefect Secret
27             which stores your Azure connection tring
28     """
29 
30     def __init__(
31         self,
32         container: str,
33         connection_string: str = None,
34         connection_string_secret: str = None,
35         **kwargs: Any
36     ) -> None:
37         self.container = container
38         self.connection_string = connection_string or os.getenv(
39             "AZURE_STORAGE_CONNECTION_STRING"
40         )
41         self.connection_string_secret = connection_string_secret
42         super().__init__(**kwargs)
43 
44     def initialize_service(self) -> None:
45         """
46         Initialize a Blob service.
47         """
48         import azure.storage.blob
49 
50         connection_string = self.connection_string
51         if not connection_string and self.connection_string_secret:
52             connection_string = Secret(self.connection_string_secret).get()
53 
54         self._service = azure.storage.blob.BlobServiceClient.from_connection_string(
55             conn_str=connection_string
56         )
57 
58     @property
59     def service(self) -> "azure.storage.blob.BlobServiceClient":
60         if not hasattr(self, "_service"):
61             self.initialize_service()
62         return self._service
63 
64     @service.setter
65     def service(self, val: Any) -> None:
66         self._service = val
67 
68     def __getstate__(self) -> dict:
69         state = self.__dict__.copy()
70         if "_service" in state:
71             del state["_service"]
72         return state
73 
74     def __setstate__(self, state: dict) -> None:
75         self.__dict__.update(state)
76 
77     def write(self, value: Any, **kwargs: Any) -> Result:
78         """
79         Writes the result value to a blob storage in Azure.
80 
81         Args:
82             - value (Any): the value to write; will then be stored as the `value` attribute
83                 of the returned `Result` instance
84             - **kwargs (optional): if provided, will be used to format the location template
85                 to determine the location to write to
86 
87         Returns:
88             - Result: a new Result instance with the appropriately formatted location
89         """
90         new = self.format(**kwargs)
91         new.value = value
92 
93         self.logger.debug("Starting to upload result to {}...".format(new.location))
94 
95         ## prepare data
96         binary_data = new.serialize_to_bytes(new.value).decode()
97 
98         # initialize client and upload
99         client = self.service.get_blob_client(
100             container=self.container, blob=new.location
101         )
102         client.upload_blob(binary_data)
103 
104         self.logger.debug("Finished uploading result to {}.".format(new.location))
105 
106         return new
107 
108     def read(self, location: str) -> Result:
109         """
110         Reads a result from an Azure Blob container and returns a corresponding `Result` instance.
111 
112         Args:
113             - location (str): the Azure blob location to read from
114 
115         Returns:
116             - Result: the read result
117         """
118         new = self.copy()
119         new.location = location
120 
121         try:
122             self.logger.debug("Starting to download result from {}...".format(location))
123 
124             # initialize client and download
125             client = self.service.get_blob_client(
126                 container=self.container, blob=location
127             )
128             content_string = client.download_blob()
129 
130             try:
131                 new.value = new.deserialize_from_bytes(content_string)
132             except EOFError:
133                 new.value = None
134             self.logger.debug("Finished downloading result from {}.".format(location))
135         except Exception as exc:
136             self.logger.exception(
137                 "Unexpected error while reading from result handler: {}".format(
138                     repr(exc)
139                 )
140             )
141             raise exc
142         return new
143 
144     def exists(self, location: str, **kwargs: Any) -> bool:
145         """
146         Checks whether the target result exists.
147 
148         Does not validate whether the result is `valid`, only that it is present.
149 
150         Args:
151             - location (str): Location of the result in the specific result target.
152                 Will check whether the provided location exists
153             - **kwargs (Any): string format arguments for `location`
154 
155         Returns:
156             - bool: whether or not the target result exists.
157         """
158         from azure.core.exceptions import ResourceNotFoundError
159 
160         # initialize client and download
161         client = self.service.get_blob_client(
162             container=self.container, blob=location.format(**kwargs)
163         )
164 
165         # Catch exception because Azure python bindings do not yet have an exists method
166         # https://github.com/Azure/azure-sdk-for-python/issues/9507
167         try:
168             client.get_blob_properties()
169             return True
170         except ResourceNotFoundError:
171             return False
172
```

Path: `src/prefect/engine/results/constant_result.py`
Content:
```
1 from typing import Any
2 from prefect.engine.result import Result
3 
4 
5 class ConstantResult(Result):
6     """
7     Hook for storing and retrieving constant Python objects. Only intended to be used
8     internally.  The "backend" in this instance is the class instance itself.
9 
10     Args:
11         - **kwargs (Any, optional): any additional `Result` initialization options
12     """
13 
14     def __init__(self, **kwargs: Any) -> None:
15         super().__init__(**kwargs)
16 
17     def read(self, location: str) -> Result:
18         """
19         Returns the underlying value regardless of the argument passed.
20 
21         Args:
22             - location (str): an unused argument
23         """
24         return self
25 
26     def write(self, value: Any, **kwargs: Any) -> Result:
27         """
28         Returns the repr of the underlying value, purely for convenience.
29 
30         Args:
31             - value (Any): unused, for interface compatibility
32             - **kwargs (optional): unused, for interface compatibility
33 
34         Raises:
35             ValueError: ConstantResults cannot be written to
36         """
37         raise ValueError("Cannot write values to `ConstantResult` types.")
38 
39     def exists(self, location: str, **kwargs: Any) -> bool:
40         """
41         As all Python objects are valid constants, always returns `True`.
42 
43         Args:
44              - location (str): for interface compatibility
45              - **kwargs (Any): string format arguments for `location`
46 
47         Returns:
48             - bool: True, confirming the constant exists.
49         """
50         return True
51
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/engine/results/azure_result.py b/src/prefect/engine/results/azure_result.py
--- a/src/prefect/engine/results/azure_result.py
+++ b/src/prefect/engine/results/azure_result.py
@@ -25,6 +25,7 @@
             will be used
         - connection_string_secret (str, optional): the name of a Prefect Secret
             which stores your Azure connection tring
+        - **kwargs (Any, optional): any additional `Result` initialization options
     """
 
     def __init__(
diff --git a/src/prefect/engine/results/constant_result.py b/src/prefect/engine/results/constant_result.py
--- a/src/prefect/engine/results/constant_result.py
+++ b/src/prefect/engine/results/constant_result.py
@@ -16,7 +16,7 @@
 
     def read(self, location: str) -> Result:
         """
-        Returns the underlying value regardless of the argument passed.
+        Will return the underlying value regardless of the argument passed.
 
         Args:
             - location (str): an unused argument
@@ -25,14 +25,14 @@
 
     def write(self, value: Any, **kwargs: Any) -> Result:
         """
-        Returns the repr of the underlying value, purely for convenience.
+        Will return the repr of the underlying value, purely for convenience.
 
         Args:
             - value (Any): unused, for interface compatibility
             - **kwargs (optional): unused, for interface compatibility
 
         Raises:
-            ValueError: ConstantResults cannot be written to
+            - ValueError: ConstantResults cannot be written to
         """
         raise ValueError("Cannot write values to `ConstantResult` types.")

{"golden_diff": "diff --git a/src/prefect/engine/results/azure_result.py b/src/prefect/engine/results/azure_result.py\n--- a/src/prefect/engine/results/azure_result.py\n+++ b/src/prefect/engine/results/azure_result.py\n@@ -25,6 +25,7 @@\n             will be used\n         - connection_string_secret (str, optional): the name of a Prefect Secret\n             which stores your Azure connection tring\n+        - **kwargs (Any, optional): any additional `Result` initialization options\n     \"\"\"\n \n     def __init__(\ndiff --git a/src/prefect/engine/results/constant_result.py b/src/prefect/engine/results/constant_result.py\n--- a/src/prefect/engine/results/constant_result.py\n+++ b/src/prefect/engine/results/constant_result.py\n@@ -16,7 +16,7 @@\n \n     def read(self, location: str) -> Result:\n         \"\"\"\n-        Returns the underlying value regardless of the argument passed.\n+        Will return the underlying value regardless of the argument passed.\n \n         Args:\n             - location (str): an unused argument\n@@ -25,14 +25,14 @@\n \n     def write(self, value: Any, **kwargs: Any) -> Result:\n         \"\"\"\n-        Returns the repr of the underlying value, purely for convenience.\n+        Will return the repr of the underlying value, purely for convenience.\n \n         Args:\n             - value (Any): unused, for interface compatibility\n             - **kwargs (optional): unused, for interface compatibility\n \n         Raises:\n-            ValueError: ConstantResults cannot be written to\n+            - ValueError: ConstantResults cannot be written to\n         \"\"\"\n         raise ValueError(\"Cannot write values to `ConstantResult` types.\")\n", "issue": "Behaviour differs from documentation: state.result & flow.get_tasks \nPrefect Version: 0.10.4\r\n\r\n## Current behavior\r\nMinor issue - The following behaviour doesn't match up to that described by the documentation on [result handling](https://docs.prefect.io/core/concepts/results.html#results):\r\n\r\n```python\r\n>>> task_ref = flow.get_tasks[0]\r\n>>> state = flow.run()\r\n>>> state.result[task_ref]._result  # a Task State's Result contains the Task's return value\r\n<Result: 1>\r\n```\r\n\r\nInstead prefect requires that the task_ref be created via a name or tag call (`flow.get_tasks(name='<name>')` instead of `flow.get_tasks[0]`) and be extracted from its list (`state.result[task_ref[0]]` instead of state.result[task_ref]):\r\n\r\n```python\r\n>>> task_ref = flow.get_tasks(name='<name>')\r\n>>> state = flow.run()\r\n>>> state.result[task_ref[0]]._result  # a Task State's Result contains the Task's return value\r\n<Result: 1>\r\n```\r\n\r\nand also the documentation on [Local Debugging](https://docs.prefect.io/core/advanced_tutorials/local-debugging.html#local-debugging):\r\n\r\n```python\r\nfrom prefect import Flow, task\r\n\r\n@task\r\ndef gotcha():\r\n    tup = ('a', ['b'])\r\n    try:\r\n        tup[1] += ['c']\r\n    except TypeError:\r\n        assert len(tup[1]) == 1\r\n\r\nflow = Flow(name=\"tuples\", tasks=[gotcha])\r\n\r\nstate = flow.run()\r\nstate.result # {<Task: gotcha>: Failed(\"Unexpected error: AssertionError()\")}\r\n\r\nfailed_state = state.result[gotcha]\r\nraise failed_state.result\r\n```\r\n\r\n## Examples\r\n\r\n### `flow.get_tasks`:\r\n\r\n```python\r\nfrom prefect import Flow, task\r\nimport pandas as pd\r\n\r\n@task(name='load data')\r\ndef _load_data() -> pd.DataFrame:\r\n    return pd.DataFrame({'a': [1, 2, 3]})\r\n\r\n\r\nwith Flow('preprocess-VO') as flow:\r\n    _load_data()\r\n\r\nstate = flow.run()\r\ntask_ref = flow.get_tasks[0]\r\n```\r\n\r\nReturns:\r\n```python\r\n>>> TypeError: 'method' object is not subscriptable\r\n```\r\n\r\n### `state.result[task_ref]`\r\n\r\n```python\r\nfrom prefect import Flow, task\r\nimport pandas as pd\r\n\r\n@task(name='load data')\r\ndef _load_data() -> pd.DataFrame:\r\n    return pd.DataFrame({'a': [1, 2, 3]})\r\n\r\n\r\nwith Flow('preprocess-VO') as flow:\r\n    _load_data()\r\n\r\nstate = flow.run()\r\ntask_ref = flow.get_tasks(name='load data')\r\nstate.result[task_ref]\r\n```\r\n\r\nReturns:\r\n```python\r\n>>> TypeError: unhashable type: 'list'\r\n```\r\n\r\n### `state.result[<function name>]`\r\n\r\n```python\r\nfrom prefect import Flow, task\r\nimport pandas as pd\r\n\r\n@task(name='load data')\r\ndef _load_data() -> pd.DataFrame:\r\n    return pd.DataFrame({'a': [1, 2, 3]})\r\n\r\n\r\nwith Flow('preprocess-VO') as flow:\r\n    _load_data()\r\n\r\nstate = flow.run()\r\ntask_ref = flow.get_tasks(name='load data')\r\nstate.result[_load_data]\r\n```\r\n\r\n```python\r\nKeyError: <Task: load data>\r\n```\r\n\r\n## Proposed behavior\r\nIt may be a bit more readable to adapt the functionality here so that Prefect behaves as in the documentation so:\r\n- `state.result[task_ref]` can be used instead of `state.result[task_ref[0]]` \r\n- `state.result[<func name>]` works\r\n- `flow.get_tasks[0]` returns a task reference\r\n\r\nAdditionally, it would be great if function names could be passed to either `state.result[<func name>]` (EDIT: I see now that this was functionality did exist) or `flow.get_tasks(<func name>)` as an alternative to adding tags and/or names to each task function to avoid having to double name all functions.\r\n\r\n---\r\n\r\nThanks for creating and maintaining prefect, really enjoying the library so far \r\n\n", "before_files": [{"content": "import os\nfrom typing import TYPE_CHECKING, Any\n\nfrom prefect.client import Secret\nfrom prefect.engine.result import Result\n\nif TYPE_CHECKING:\n    import azure.storage.blob\n\n\nclass AzureResult(Result):\n    \"\"\"\n    Result for writing to and reading from an Azure Blob storage.\n\n    Note that your flow's runtime environment must be able to authenticate with\n    Azure; there are currently two supported options: provide a connection string\n    either at initialization or at runtime through an environment variable, or\n    set your Azure connection string as a Prefect Secret.  Using an environment\n    variable is the recommended approach.\n\n    Args:\n        - container (str): the name of the container to write to / read from\n        - connection_string (str, optional): an Azure connection string for communicating with\n            Blob storage. If not provided the value set in the environment as `AZURE_STORAGE_CONNECTION_STRING`\n            will be used\n        - connection_string_secret (str, optional): the name of a Prefect Secret\n            which stores your Azure connection tring\n    \"\"\"\n\n    def __init__(\n        self,\n        container: str,\n        connection_string: str = None,\n        connection_string_secret: str = None,\n        **kwargs: Any\n    ) -> None:\n        self.container = container\n        self.connection_string = connection_string or os.getenv(\n            \"AZURE_STORAGE_CONNECTION_STRING\"\n        )\n        self.connection_string_secret = connection_string_secret\n        super().__init__(**kwargs)\n\n    def initialize_service(self) -> None:\n        \"\"\"\n        Initialize a Blob service.\n        \"\"\"\n        import azure.storage.blob\n\n        connection_string = self.connection_string\n        if not connection_string and self.connection_string_secret:\n            connection_string = Secret(self.connection_string_secret).get()\n\n        self._service = azure.storage.blob.BlobServiceClient.from_connection_string(\n            conn_str=connection_string\n        )\n\n    @property\n    def service(self) -> \"azure.storage.blob.BlobServiceClient\":\n        if not hasattr(self, \"_service\"):\n            self.initialize_service()\n        return self._service\n\n    @service.setter\n    def service(self, val: Any) -> None:\n        self._service = val\n\n    def __getstate__(self) -> dict:\n        state = self.__dict__.copy()\n        if \"_service\" in state:\n            del state[\"_service\"]\n        return state\n\n    def __setstate__(self, state: dict) -> None:\n        self.__dict__.update(state)\n\n    def write(self, value: Any, **kwargs: Any) -> Result:\n        \"\"\"\n        Writes the result value to a blob storage in Azure.\n\n        Args:\n            - value (Any): the value to write; will then be stored as the `value` attribute\n                of the returned `Result` instance\n            - **kwargs (optional): if provided, will be used to format the location template\n                to determine the location to write to\n\n        Returns:\n            - Result: a new Result instance with the appropriately formatted location\n        \"\"\"\n        new = self.format(**kwargs)\n        new.value = value\n\n        self.logger.debug(\"Starting to upload result to {}...\".format(new.location))\n\n        ## prepare data\n        binary_data = new.serialize_to_bytes(new.value).decode()\n\n        # initialize client and upload\n        client = self.service.get_blob_client(\n            container=self.container, blob=new.location\n        )\n        client.upload_blob(binary_data)\n\n        self.logger.debug(\"Finished uploading result to {}.\".format(new.location))\n\n        return new\n\n    def read(self, location: str) -> Result:\n        \"\"\"\n        Reads a result from an Azure Blob container and returns a corresponding `Result` instance.\n\n        Args:\n            - location (str): the Azure blob location to read from\n\n        Returns:\n            - Result: the read result\n        \"\"\"\n        new = self.copy()\n        new.location = location\n\n        try:\n            self.logger.debug(\"Starting to download result from {}...\".format(location))\n\n            # initialize client and download\n            client = self.service.get_blob_client(\n                container=self.container, blob=location\n            )\n            content_string = client.download_blob()\n\n            try:\n                new.value = new.deserialize_from_bytes(content_string)\n            except EOFError:\n                new.value = None\n            self.logger.debug(\"Finished downloading result from {}.\".format(location))\n        except Exception as exc:\n            self.logger.exception(\n                \"Unexpected error while reading from result handler: {}\".format(\n                    repr(exc)\n                )\n            )\n            raise exc\n        return new\n\n    def exists(self, location: str, **kwargs: Any) -> bool:\n        \"\"\"\n        Checks whether the target result exists.\n\n        Does not validate whether the result is `valid`, only that it is present.\n\n        Args:\n            - location (str): Location of the result in the specific result target.\n                Will check whether the provided location exists\n            - **kwargs (Any): string format arguments for `location`\n\n        Returns:\n            - bool: whether or not the target result exists.\n        \"\"\"\n        from azure.core.exceptions import ResourceNotFoundError\n\n        # initialize client and download\n        client = self.service.get_blob_client(\n            container=self.container, blob=location.format(**kwargs)\n        )\n\n        # Catch exception because Azure python bindings do not yet have an exists method\n        # https://github.com/Azure/azure-sdk-for-python/issues/9507\n        try:\n            client.get_blob_properties()\n            return True\n        except ResourceNotFoundError:\n            return False\n", "path": "src/prefect/engine/results/azure_result.py"}, {"content": "from typing import Any\nfrom prefect.engine.result import Result\n\n\nclass ConstantResult(Result):\n    \"\"\"\n    Hook for storing and retrieving constant Python objects. Only intended to be used\n    internally.  The \"backend\" in this instance is the class instance itself.\n\n    Args:\n        - **kwargs (Any, optional): any additional `Result` initialization options\n    \"\"\"\n\n    def __init__(self, **kwargs: Any) -> None:\n        super().__init__(**kwargs)\n\n    def read(self, location: str) -> Result:\n        \"\"\"\n        Returns the underlying value regardless of the argument passed.\n\n        Args:\n            - location (str): an unused argument\n        \"\"\"\n        return self\n\n    def write(self, value: Any, **kwargs: Any) -> Result:\n        \"\"\"\n        Returns the repr of the underlying value, purely for convenience.\n\n        Args:\n            - value (Any): unused, for interface compatibility\n            - **kwargs (optional): unused, for interface compatibility\n\n        Raises:\n            ValueError: ConstantResults cannot be written to\n        \"\"\"\n        raise ValueError(\"Cannot write values to `ConstantResult` types.\")\n\n    def exists(self, location: str, **kwargs: Any) -> bool:\n        \"\"\"\n        As all Python objects are valid constants, always returns `True`.\n\n        Args:\n             - location (str): for interface compatibility\n             - **kwargs (Any): string format arguments for `location`\n\n        Returns:\n            - bool: True, confirming the constant exists.\n        \"\"\"\n        return True\n", "path": "src/prefect/engine/results/constant_result.py"}], "after_files": [{"content": "import os\nfrom typing import TYPE_CHECKING, Any\n\nfrom prefect.client import Secret\nfrom prefect.engine.result import Result\n\nif TYPE_CHECKING:\n    import azure.storage.blob\n\n\nclass AzureResult(Result):\n    \"\"\"\n    Result for writing to and reading from an Azure Blob storage.\n\n    Note that your flow's runtime environment must be able to authenticate with\n    Azure; there are currently two supported options: provide a connection string\n    either at initialization or at runtime through an environment variable, or\n    set your Azure connection string as a Prefect Secret.  Using an environment\n    variable is the recommended approach.\n\n    Args:\n        - container (str): the name of the container to write to / read from\n        - connection_string (str, optional): an Azure connection string for communicating with\n            Blob storage. If not provided the value set in the environment as `AZURE_STORAGE_CONNECTION_STRING`\n            will be used\n        - connection_string_secret (str, optional): the name of a Prefect Secret\n            which stores your Azure connection tring\n        - **kwargs (Any, optional): any additional `Result` initialization options\n    \"\"\"\n\n    def __init__(\n        self,\n        container: str,\n        connection_string: str = None,\n        connection_string_secret: str = None,\n        **kwargs: Any\n    ) -> None:\n        self.container = container\n        self.connection_string = connection_string or os.getenv(\n            \"AZURE_STORAGE_CONNECTION_STRING\"\n        )\n        self.connection_string_secret = connection_string_secret\n        super().__init__(**kwargs)\n\n    def initialize_service(self) -> None:\n        \"\"\"\n        Initialize a Blob service.\n        \"\"\"\n        import azure.storage.blob\n\n        connection_string = self.connection_string\n        if not connection_string and self.connection_string_secret:\n            connection_string = Secret(self.connection_string_secret).get()\n\n        self._service = azure.storage.blob.BlobServiceClient.from_connection_string(\n            conn_str=connection_string\n        )\n\n    @property\n    def service(self) -> \"azure.storage.blob.BlobServiceClient\":\n        if not hasattr(self, \"_service\"):\n            self.initialize_service()\n        return self._service\n\n    @service.setter\n    def service(self, val: Any) -> None:\n        self._service = val\n\n    def __getstate__(self) -> dict:\n        state = self.__dict__.copy()\n        if \"_service\" in state:\n            del state[\"_service\"]\n        return state\n\n    def __setstate__(self, state: dict) -> None:\n        self.__dict__.update(state)\n\n    def write(self, value: Any, **kwargs: Any) -> Result:\n        \"\"\"\n        Writes the result value to a blob storage in Azure.\n\n        Args:\n            - value (Any): the value to write; will then be stored as the `value` attribute\n                of the returned `Result` instance\n            - **kwargs (optional): if provided, will be used to format the location template\n                to determine the location to write to\n\n        Returns:\n            - Result: a new Result instance with the appropriately formatted location\n        \"\"\"\n        new = self.format(**kwargs)\n        new.value = value\n\n        self.logger.debug(\"Starting to upload result to {}...\".format(new.location))\n\n        ## prepare data\n        binary_data = new.serialize_to_bytes(new.value).decode()\n\n        # initialize client and upload\n        client = self.service.get_blob_client(\n            container=self.container, blob=new.location\n        )\n        client.upload_blob(binary_data)\n\n        self.logger.debug(\"Finished uploading result to {}.\".format(new.location))\n\n        return new\n\n    def read(self, location: str) -> Result:\n        \"\"\"\n        Reads a result from an Azure Blob container and returns a corresponding `Result` instance.\n\n        Args:\n            - location (str): the Azure blob location to read from\n\n        Returns:\n            - Result: the read result\n        \"\"\"\n        new = self.copy()\n        new.location = location\n\n        try:\n            self.logger.debug(\"Starting to download result from {}...\".format(location))\n\n            # initialize client and download\n            client = self.service.get_blob_client(\n                container=self.container, blob=location\n            )\n            content_string = client.download_blob()\n\n            try:\n                new.value = new.deserialize_from_bytes(content_string)\n            except EOFError:\n                new.value = None\n            self.logger.debug(\"Finished downloading result from {}.\".format(location))\n        except Exception as exc:\n            self.logger.exception(\n                \"Unexpected error while reading from result handler: {}\".format(\n                    repr(exc)\n                )\n            )\n            raise exc\n        return new\n\n    def exists(self, location: str, **kwargs: Any) -> bool:\n        \"\"\"\n        Checks whether the target result exists.\n\n        Does not validate whether the result is `valid`, only that it is present.\n\n        Args:\n            - location (str): Location of the result in the specific result target.\n                Will check whether the provided location exists\n            - **kwargs (Any): string format arguments for `location`\n\n        Returns:\n            - bool: whether or not the target result exists.\n        \"\"\"\n        from azure.core.exceptions import ResourceNotFoundError\n\n        # initialize client and download\n        client = self.service.get_blob_client(\n            container=self.container, blob=location.format(**kwargs)\n        )\n\n        # Catch exception because Azure python bindings do not yet have an exists method\n        # https://github.com/Azure/azure-sdk-for-python/issues/9507\n        try:\n            client.get_blob_properties()\n            return True\n        except ResourceNotFoundError:\n            return False\n", "path": "src/prefect/engine/results/azure_result.py"}, {"content": "from typing import Any\nfrom prefect.engine.result import Result\n\n\nclass ConstantResult(Result):\n    \"\"\"\n    Hook for storing and retrieving constant Python objects. Only intended to be used\n    internally.  The \"backend\" in this instance is the class instance itself.\n\n    Args:\n        - **kwargs (Any, optional): any additional `Result` initialization options\n    \"\"\"\n\n    def __init__(self, **kwargs: Any) -> None:\n        super().__init__(**kwargs)\n\n    def read(self, location: str) -> Result:\n        \"\"\"\n        Will return the underlying value regardless of the argument passed.\n\n        Args:\n            - location (str): an unused argument\n        \"\"\"\n        return self\n\n    def write(self, value: Any, **kwargs: Any) -> Result:\n        \"\"\"\n        Will return the repr of the underlying value, purely for convenience.\n\n        Args:\n            - value (Any): unused, for interface compatibility\n            - **kwargs (optional): unused, for interface compatibility\n\n        Raises:\n            - ValueError: ConstantResults cannot be written to\n        \"\"\"\n        raise ValueError(\"Cannot write values to `ConstantResult` types.\")\n\n    def exists(self, location: str, **kwargs: Any) -> bool:\n        \"\"\"\n        As all Python objects are valid constants, always returns `True`.\n\n        Args:\n             - location (str): for interface compatibility\n             - **kwargs (Any): string format arguments for `location`\n\n        Returns:\n            - bool: True, confirming the constant exists.\n        \"\"\"\n        return True\n", "path": "src/prefect/engine/results/constant_result.py"}]}