We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
AzureResult read function not working
## Description
<!-- A clear description of the bug -->
Using `AzureResult` to read a file, an exception is raised:

> TypeError: a bytes-like object is required, not 'StorageStreamDownloader'

Looking at the `read` function in `src/prefect/engine/results/azure_result.py` I see the following:

```python
content_string = client.download_blob()

try:
    new.value = new.serializer.deserialize(content_string)
```

`client.download_blob()` returns a `StorageStreamDownloader`, while `new.serailizer.deserialize(content_string)` expects a byte array.

## Expected Behavior
I would expect that the file would be read without raising an exception.

## Reproduction
```python
from prefect.engine.results import AzureResult

con_string = ""
azure_result = AzureResult(container="data", connection_string=con_string)

w = azure_result.write({"Test": "123"})
r = azure_result.read(w.location)
```


## Environment
```json
{
  "config_overrides": {
    "context": {
      "secrets": false
    }
  },
  "env_vars": [],
  "system_information": {
    "platform": "Windows-10-10.0.18362-SP0",
    "prefect_backend": "server",
    "prefect_version": "0.13.19",
    "python_version": "3.9.0"
  }
}
```

--- 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
25             `AZURE_STORAGE_CONNECTION_STRING` will be used
26         - connection_string_secret (str, optional): the name of a Prefect Secret
27             which stores your Azure connection tring
28         - **kwargs (Any, optional): any additional `Result` initialization options
29     """
30 
31     def __init__(
32         self,
33         container: str,
34         connection_string: str = None,
35         connection_string_secret: str = None,
36         **kwargs: Any
37     ) -> None:
38         self.container = container
39         self.connection_string = connection_string or os.getenv(
40             "AZURE_STORAGE_CONNECTION_STRING"
41         )
42         self.connection_string_secret = connection_string_secret
43         super().__init__(**kwargs)
44 
45     def initialize_service(self) -> None:
46         """
47         Initialize a Blob service.
48         """
49         import azure.storage.blob
50 
51         connection_string = self.connection_string
52         if not connection_string and self.connection_string_secret:
53             connection_string = Secret(self.connection_string_secret).get()
54 
55         self._service = azure.storage.blob.BlobServiceClient.from_connection_string(
56             conn_str=connection_string
57         )
58 
59     @property
60     def service(self) -> "azure.storage.blob.BlobServiceClient":
61         if not hasattr(self, "_service"):
62             self.initialize_service()
63         return self._service
64 
65     @service.setter
66     def service(self, val: Any) -> None:
67         self._service = val
68 
69     def __getstate__(self) -> dict:
70         state = self.__dict__.copy()
71         if "_service" in state:
72             del state["_service"]
73         return state
74 
75     def __setstate__(self, state: dict) -> None:
76         self.__dict__.update(state)
77 
78     def write(self, value_: Any, **kwargs: Any) -> Result:
79         """
80         Writes the result value to a blob storage in Azure.
81 
82         Args:
83             - value_ (Any): the value to write; will then be stored as the `value` attribute
84                 of the returned `Result` instance
85             - **kwargs (optional): if provided, will be used to format the location template
86                 to determine the location to write to
87 
88         Returns:
89             - Result: a new Result instance with the appropriately formatted location
90         """
91         new = self.format(**kwargs)
92         new.value = value_
93 
94         self.logger.debug("Starting to upload result to {}...".format(new.location))
95 
96         # prepare data
97         binary_data = new.serializer.serialize(new.value)
98 
99         # initialize client and upload
100         client = self.service.get_blob_client(
101             container=self.container, blob=new.location
102         )
103         client.upload_blob(binary_data)
104 
105         self.logger.debug("Finished uploading result to {}.".format(new.location))
106 
107         return new
108 
109     def read(self, location: str) -> Result:
110         """
111         Reads a result from an Azure Blob container and returns a corresponding `Result` instance.
112 
113         Args:
114             - location (str): the Azure blob location to read from
115 
116         Returns:
117             - Result: the read result
118         """
119         new = self.copy()
120         new.location = location
121 
122         try:
123             self.logger.debug("Starting to download result from {}...".format(location))
124 
125             # initialize client and download
126             client = self.service.get_blob_client(
127                 container=self.container, blob=location
128             )
129             content_string = client.download_blob()
130 
131             try:
132                 new.value = new.serializer.deserialize(content_string)
133             except EOFError:
134                 new.value = None
135             self.logger.debug("Finished downloading result from {}.".format(location))
136         except Exception as exc:
137             self.logger.exception(
138                 "Unexpected error while reading from result handler: {}".format(
139                     repr(exc)
140                 )
141             )
142             raise exc
143         return new
144 
145     def exists(self, location: str, **kwargs: Any) -> bool:
146         """
147         Checks whether the target result exists.
148 
149         Does not validate whether the result is `valid`, only that it is present.
150 
151         Args:
152             - location (str): Location of the result in the specific result target.
153                 Will check whether the provided location exists
154             - **kwargs (Any): string format arguments for `location`
155 
156         Returns:
157             - bool: whether or not the target result exists.
158         """
159         from azure.core.exceptions import ResourceNotFoundError
160 
161         # initialize client and download
162         client = self.service.get_blob_client(
163             container=self.container, blob=location.format(**kwargs)
164         )
165 
166         # Catch exception because Azure python bindings do not yet have an exists method
167         # https://github.com/Azure/azure-sdk-for-python/issues/9507
168         try:
169             client.get_blob_properties()
170             return True
171         except ResourceNotFoundError:
172             return False
173
```
--- 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
@@ -126,10 +126,10 @@
             client = self.service.get_blob_client(
                 container=self.container, blob=location
             )
-            content_string = client.download_blob()
+            content_bytes = client.download_blob().content_as_bytes()
 
             try:
-                new.value = new.serializer.deserialize(content_string)
+                new.value = new.serializer.deserialize(content_bytes)
             except EOFError:
                 new.value = None
             self.logger.debug("Finished downloading result from {}.".format(location))

{"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@@ -126,10 +126,10 @@\n             client = self.service.get_blob_client(\n                 container=self.container, blob=location\n             )\n-            content_string = client.download_blob()\n+            content_bytes = client.download_blob().content_as_bytes()\n \n             try:\n-                new.value = new.serializer.deserialize(content_string)\n+                new.value = new.serializer.deserialize(content_bytes)\n             except EOFError:\n                 new.value = None\n             self.logger.debug(\"Finished downloading result from {}.\".format(location))\n", "issue": "AzureResult read function not working\n## Description\r\n<!-- A clear description of the bug -->\r\nUsing `AzureResult` to read a file, an exception is raised:\r\n\r\n> TypeError: a bytes-like object is required, not 'StorageStreamDownloader'\r\n\r\nLooking at the `read` function in `src/prefect/engine/results/azure_result.py` I see the following:\r\n\r\n```python\r\ncontent_string = client.download_blob()\r\n\r\ntry:\r\n    new.value = new.serializer.deserialize(content_string)\r\n```\r\n\r\n`client.download_blob()` returns a `StorageStreamDownloader`, while `new.serailizer.deserialize(content_string)` expects a byte array.\r\n\r\n## Expected Behavior\r\nI would expect that the file would be read without raising an exception.\r\n\r\n## Reproduction\r\n```python\r\nfrom prefect.engine.results import AzureResult\r\n\r\ncon_string = \"\"\r\nazure_result = AzureResult(container=\"data\", connection_string=con_string)\r\n\r\nw = azure_result.write({\"Test\": \"123\"})\r\nr = azure_result.read(w.location)\r\n```\r\n\r\n\r\n## Environment\r\n```json\r\n{\r\n  \"config_overrides\": {\r\n    \"context\": {\r\n      \"secrets\": false\r\n    }\r\n  },\r\n  \"env_vars\": [],\r\n  \"system_information\": {\r\n    \"platform\": \"Windows-10-10.0.18362-SP0\",\r\n    \"prefect_backend\": \"server\",\r\n    \"prefect_version\": \"0.13.19\",\r\n    \"python_version\": \"3.9.0\"\r\n  }\r\n}\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\n            `AZURE_STORAGE_CONNECTION_STRING` 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.serializer.serialize(new.value)\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.serializer.deserialize(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"}], "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\n            `AZURE_STORAGE_CONNECTION_STRING` 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.serializer.serialize(new.value)\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_bytes = client.download_blob().content_as_bytes()\n\n            try:\n                new.value = new.serializer.deserialize(content_bytes)\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"}]}