We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Azure BlobStorageUpload doesn't allow for overwriting blobs
## Current behavior

You get an error if you try to upload the same file name

```
azure.core.exceptions.ResourceExistsError: The specified blob already exists.
RequestId:5bef0cf1-b01e-002e-6
```

## Proposed behavior

The task should take in an `overwrite` argument and pass it to [this line](https://github.com/PrefectHQ/prefect/blob/6cd24b023411980842fa77e6c0ca2ced47eeb83e/src/prefect/tasks/azure/blobstorage.py#L131).



--- 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/tasks/azure/blobstorage.py`
Content:
```
1 import uuid
2 
3 import azure.storage.blob
4 
5 from prefect import Task
6 from prefect.client import Secret
7 from prefect.utilities.tasks import defaults_from_attrs
8 
9 
10 class BlobStorageDownload(Task):
11     """
12     Task for downloading data from an Blob Storage container and returning it as a string.
13     Note that all initialization arguments can optionally be provided or overwritten at runtime.
14 
15     Args:
16         - azure_credentials_secret (str, optional): the name of the Prefect Secret
17             that stores your Azure credentials; this Secret must be an Azure connection string
18         - container (str, optional): the name of the Azure Blob Storage to download from
19         - **kwargs (dict, optional): additional keyword arguments to pass to the
20             Task constructor
21     """
22 
23     def __init__(
24         self,
25         azure_credentials_secret: str = "AZ_CONNECTION_STRING",
26         container: str = None,
27         **kwargs
28     ) -> None:
29         self.azure_credentials_secret = azure_credentials_secret
30         self.container = container
31         super().__init__(**kwargs)
32 
33     @defaults_from_attrs("azure_credentials_secret", "container")
34     def run(
35         self,
36         blob_name: str,
37         azure_credentials_secret: str = "AZ_CONNECTION_STRING",
38         container: str = None,
39     ) -> str:
40         """
41         Task run method.
42 
43         Args:
44             - blob_name (str): the name of the blob within this container to retrieve
45             - azure_credentials_secret (str, optional): the name of the Prefect Secret
46             that stores your Azure credentials; this Secret must be an Azure connection string
47             - container (str, optional): the name of the Blob Storage container to download from
48 
49         Returns:
50             - str: the contents of this blob_name / container, as a string
51         """
52 
53         if container is None:
54             raise ValueError("A container name must be provided.")
55 
56         # get Azure credentials
57         azure_credentials = Secret(azure_credentials_secret).get()
58 
59         blob_service = azure.storage.blob.BlobServiceClient.from_connection_string(
60             conn_str=azure_credentials
61         )
62 
63         client = blob_service.get_blob_client(container=container, blob=blob_name)
64         content_string = client.download_blob().content_as_text()
65 
66         return content_string
67 
68 
69 class BlobStorageUpload(Task):
70     """
71     Task for uploading string data (e.g., a JSON string) to an Azure Blob Storage container.
72     Note that all initialization arguments can optionally be provided or overwritten at runtime.
73 
74     Args:
75         - azure_credentials_secret (str, optional): the name of the Prefect Secret
76             that stores your Azure credentials; this Secret must be an Azure connection string
77         - container (str, optional): the name of the Azure Blob Storage to upload to
78         - **kwargs (dict, optional): additional keyword arguments to pass to the
79             Task constructor
80     """
81 
82     def __init__(
83         self,
84         azure_credentials_secret: str = "AZ_CONNECTION_STRING",
85         container: str = None,
86         **kwargs
87     ) -> None:
88         self.azure_credentials_secret = azure_credentials_secret
89         self.container = container
90         super().__init__(**kwargs)
91 
92     @defaults_from_attrs("azure_credentials_secret", "container")
93     def run(
94         self,
95         data: str,
96         blob_name: str = None,
97         azure_credentials_secret: str = "AZ_CONNECTION_STRING",
98         container: str = None,
99     ) -> str:
100         """
101         Task run method.
102 
103         Args:
104             - data (str): the data payload to upload
105             - blob_name (str, optional): the name to upload the data under; if not
106                     provided, a random `uuid` will be created
107             - azure_credentials_secret (str, optional): the name of the Prefect Secret
108             that stores your Azure credentials; this Secret must be an Azure connection string
109             - container (str, optional): the name of the Blob Storage container to upload to
110 
111         Returns:
112             - str: the name of the blob the data payload was uploaded to
113         """
114 
115         if container is None:
116             raise ValueError("A container name must be provided.")
117 
118         # get Azure credentials
119         azure_credentials = Secret(azure_credentials_secret).get()
120 
121         blob_service = azure.storage.blob.BlobServiceClient.from_connection_string(
122             conn_str=azure_credentials
123         )
124 
125         # create key if not provided
126         if blob_name is None:
127             blob_name = str(uuid.uuid4())
128 
129         client = blob_service.get_blob_client(container=container, blob=blob_name)
130 
131         client.upload_blob(data)
132 
133         return blob_name
134
```
--- 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/tasks/azure/blobstorage.py b/src/prefect/tasks/azure/blobstorage.py
--- a/src/prefect/tasks/azure/blobstorage.py
+++ b/src/prefect/tasks/azure/blobstorage.py
@@ -75,6 +75,8 @@
         - azure_credentials_secret (str, optional): the name of the Prefect Secret
             that stores your Azure credentials; this Secret must be an Azure connection string
         - container (str, optional): the name of the Azure Blob Storage to upload to
+        - overwrite (bool, optional): if `True`, an existing blob with the same name will be overwritten.
+            Defaults to `False` and an error will be thrown if the blob already exists.
         - **kwargs (dict, optional): additional keyword arguments to pass to the
             Task constructor
     """
@@ -83,19 +85,22 @@
         self,
         azure_credentials_secret: str = "AZ_CONNECTION_STRING",
         container: str = None,
+        overwrite: bool = False,
         **kwargs
     ) -> None:
         self.azure_credentials_secret = azure_credentials_secret
         self.container = container
+        self.overwrite = overwrite
         super().__init__(**kwargs)
 
-    @defaults_from_attrs("azure_credentials_secret", "container")
+    @defaults_from_attrs("azure_credentials_secret", "container", "overwrite")
     def run(
         self,
         data: str,
         blob_name: str = None,
         azure_credentials_secret: str = "AZ_CONNECTION_STRING",
         container: str = None,
+        overwrite: bool = False,
     ) -> str:
         """
         Task run method.
@@ -107,6 +112,8 @@
             - azure_credentials_secret (str, optional): the name of the Prefect Secret
             that stores your Azure credentials; this Secret must be an Azure connection string
             - container (str, optional): the name of the Blob Storage container to upload to
+        - overwrite (bool, optional): if `True`, an existing blob with the same name will be overwritten.
+            Defaults to `False` and an error will be thrown if the blob already exists.
 
         Returns:
             - str: the name of the blob the data payload was uploaded to
@@ -128,6 +135,6 @@
 
         client = blob_service.get_blob_client(container=container, blob=blob_name)
 
-        client.upload_blob(data)
+        client.upload_blob(data, overwrite=overwrite)
 
         return blob_name

{"golden_diff": "diff --git a/src/prefect/tasks/azure/blobstorage.py b/src/prefect/tasks/azure/blobstorage.py\n--- a/src/prefect/tasks/azure/blobstorage.py\n+++ b/src/prefect/tasks/azure/blobstorage.py\n@@ -75,6 +75,8 @@\n         - azure_credentials_secret (str, optional): the name of the Prefect Secret\n             that stores your Azure credentials; this Secret must be an Azure connection string\n         - container (str, optional): the name of the Azure Blob Storage to upload to\n+        - overwrite (bool, optional): if `True`, an existing blob with the same name will be overwritten.\n+            Defaults to `False` and an error will be thrown if the blob already exists.\n         - **kwargs (dict, optional): additional keyword arguments to pass to the\n             Task constructor\n     \"\"\"\n@@ -83,19 +85,22 @@\n         self,\n         azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n         container: str = None,\n+        overwrite: bool = False,\n         **kwargs\n     ) -> None:\n         self.azure_credentials_secret = azure_credentials_secret\n         self.container = container\n+        self.overwrite = overwrite\n         super().__init__(**kwargs)\n \n-    @defaults_from_attrs(\"azure_credentials_secret\", \"container\")\n+    @defaults_from_attrs(\"azure_credentials_secret\", \"container\", \"overwrite\")\n     def run(\n         self,\n         data: str,\n         blob_name: str = None,\n         azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n         container: str = None,\n+        overwrite: bool = False,\n     ) -> str:\n         \"\"\"\n         Task run method.\n@@ -107,6 +112,8 @@\n             - azure_credentials_secret (str, optional): the name of the Prefect Secret\n             that stores your Azure credentials; this Secret must be an Azure connection string\n             - container (str, optional): the name of the Blob Storage container to upload to\n+        - overwrite (bool, optional): if `True`, an existing blob with the same name will be overwritten.\n+            Defaults to `False` and an error will be thrown if the blob already exists.\n \n         Returns:\n             - str: the name of the blob the data payload was uploaded to\n@@ -128,6 +135,6 @@\n \n         client = blob_service.get_blob_client(container=container, blob=blob_name)\n \n-        client.upload_blob(data)\n+        client.upload_blob(data, overwrite=overwrite)\n \n         return blob_name\n", "issue": "Azure BlobStorageUpload doesn't allow for overwriting blobs\n## Current behavior\r\n\r\nYou get an error if you try to upload the same file name\r\n\r\n```\r\nazure.core.exceptions.ResourceExistsError: The specified blob already exists.\r\nRequestId:5bef0cf1-b01e-002e-6\r\n```\r\n\r\n## Proposed behavior\r\n\r\nThe task should take in an `overwrite` argument and pass it to [this line](https://github.com/PrefectHQ/prefect/blob/6cd24b023411980842fa77e6c0ca2ced47eeb83e/src/prefect/tasks/azure/blobstorage.py#L131).\r\n\r\n\n", "before_files": [{"content": "import uuid\n\nimport azure.storage.blob\n\nfrom prefect import Task\nfrom prefect.client import Secret\nfrom prefect.utilities.tasks import defaults_from_attrs\n\n\nclass BlobStorageDownload(Task):\n    \"\"\"\n    Task for downloading data from an Blob Storage container and returning it as a string.\n    Note that all initialization arguments can optionally be provided or overwritten at runtime.\n\n    Args:\n        - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n        - container (str, optional): the name of the Azure Blob Storage to download from\n        - **kwargs (dict, optional): additional keyword arguments to pass to the\n            Task constructor\n    \"\"\"\n\n    def __init__(\n        self,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n        **kwargs\n    ) -> None:\n        self.azure_credentials_secret = azure_credentials_secret\n        self.container = container\n        super().__init__(**kwargs)\n\n    @defaults_from_attrs(\"azure_credentials_secret\", \"container\")\n    def run(\n        self,\n        blob_name: str,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n    ) -> str:\n        \"\"\"\n        Task run method.\n\n        Args:\n            - blob_name (str): the name of the blob within this container to retrieve\n            - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n            - container (str, optional): the name of the Blob Storage container to download from\n\n        Returns:\n            - str: the contents of this blob_name / container, as a string\n        \"\"\"\n\n        if container is None:\n            raise ValueError(\"A container name must be provided.\")\n\n        # get Azure credentials\n        azure_credentials = Secret(azure_credentials_secret).get()\n\n        blob_service = azure.storage.blob.BlobServiceClient.from_connection_string(\n            conn_str=azure_credentials\n        )\n\n        client = blob_service.get_blob_client(container=container, blob=blob_name)\n        content_string = client.download_blob().content_as_text()\n\n        return content_string\n\n\nclass BlobStorageUpload(Task):\n    \"\"\"\n    Task for uploading string data (e.g., a JSON string) to an Azure Blob Storage container.\n    Note that all initialization arguments can optionally be provided or overwritten at runtime.\n\n    Args:\n        - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n        - container (str, optional): the name of the Azure Blob Storage to upload to\n        - **kwargs (dict, optional): additional keyword arguments to pass to the\n            Task constructor\n    \"\"\"\n\n    def __init__(\n        self,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n        **kwargs\n    ) -> None:\n        self.azure_credentials_secret = azure_credentials_secret\n        self.container = container\n        super().__init__(**kwargs)\n\n    @defaults_from_attrs(\"azure_credentials_secret\", \"container\")\n    def run(\n        self,\n        data: str,\n        blob_name: str = None,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n    ) -> str:\n        \"\"\"\n        Task run method.\n\n        Args:\n            - data (str): the data payload to upload\n            - blob_name (str, optional): the name to upload the data under; if not\n                    provided, a random `uuid` will be created\n            - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n            - container (str, optional): the name of the Blob Storage container to upload to\n\n        Returns:\n            - str: the name of the blob the data payload was uploaded to\n        \"\"\"\n\n        if container is None:\n            raise ValueError(\"A container name must be provided.\")\n\n        # get Azure credentials\n        azure_credentials = Secret(azure_credentials_secret).get()\n\n        blob_service = azure.storage.blob.BlobServiceClient.from_connection_string(\n            conn_str=azure_credentials\n        )\n\n        # create key if not provided\n        if blob_name is None:\n            blob_name = str(uuid.uuid4())\n\n        client = blob_service.get_blob_client(container=container, blob=blob_name)\n\n        client.upload_blob(data)\n\n        return blob_name\n", "path": "src/prefect/tasks/azure/blobstorage.py"}], "after_files": [{"content": "import uuid\n\nimport azure.storage.blob\n\nfrom prefect import Task\nfrom prefect.client import Secret\nfrom prefect.utilities.tasks import defaults_from_attrs\n\n\nclass BlobStorageDownload(Task):\n    \"\"\"\n    Task for downloading data from an Blob Storage container and returning it as a string.\n    Note that all initialization arguments can optionally be provided or overwritten at runtime.\n\n    Args:\n        - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n        - container (str, optional): the name of the Azure Blob Storage to download from\n        - **kwargs (dict, optional): additional keyword arguments to pass to the\n            Task constructor\n    \"\"\"\n\n    def __init__(\n        self,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n        **kwargs\n    ) -> None:\n        self.azure_credentials_secret = azure_credentials_secret\n        self.container = container\n        super().__init__(**kwargs)\n\n    @defaults_from_attrs(\"azure_credentials_secret\", \"container\")\n    def run(\n        self,\n        blob_name: str,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n    ) -> str:\n        \"\"\"\n        Task run method.\n\n        Args:\n            - blob_name (str): the name of the blob within this container to retrieve\n            - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n            - container (str, optional): the name of the Blob Storage container to download from\n\n        Returns:\n            - str: the contents of this blob_name / container, as a string\n        \"\"\"\n\n        if container is None:\n            raise ValueError(\"A container name must be provided.\")\n\n        # get Azure credentials\n        azure_credentials = Secret(azure_credentials_secret).get()\n\n        blob_service = azure.storage.blob.BlobServiceClient.from_connection_string(\n            conn_str=azure_credentials\n        )\n\n        client = blob_service.get_blob_client(container=container, blob=blob_name)\n        content_string = client.download_blob().content_as_text()\n\n        return content_string\n\n\nclass BlobStorageUpload(Task):\n    \"\"\"\n    Task for uploading string data (e.g., a JSON string) to an Azure Blob Storage container.\n    Note that all initialization arguments can optionally be provided or overwritten at runtime.\n\n    Args:\n        - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n        - container (str, optional): the name of the Azure Blob Storage to upload to\n        - overwrite (bool, optional): if `True`, an existing blob with the same name will be overwritten.\n            Defaults to `False` and an error will be thrown if the blob already exists.\n        - **kwargs (dict, optional): additional keyword arguments to pass to the\n            Task constructor\n    \"\"\"\n\n    def __init__(\n        self,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n        overwrite: bool = False,\n        **kwargs\n    ) -> None:\n        self.azure_credentials_secret = azure_credentials_secret\n        self.container = container\n        self.overwrite = overwrite\n        super().__init__(**kwargs)\n\n    @defaults_from_attrs(\"azure_credentials_secret\", \"container\", \"overwrite\")\n    def run(\n        self,\n        data: str,\n        blob_name: str = None,\n        azure_credentials_secret: str = \"AZ_CONNECTION_STRING\",\n        container: str = None,\n        overwrite: bool = False,\n    ) -> str:\n        \"\"\"\n        Task run method.\n\n        Args:\n            - data (str): the data payload to upload\n            - blob_name (str, optional): the name to upload the data under; if not\n                    provided, a random `uuid` will be created\n            - azure_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your Azure credentials; this Secret must be an Azure connection string\n            - container (str, optional): the name of the Blob Storage container to upload to\n        - overwrite (bool, optional): if `True`, an existing blob with the same name will be overwritten.\n            Defaults to `False` and an error will be thrown if the blob already exists.\n\n        Returns:\n            - str: the name of the blob the data payload was uploaded to\n        \"\"\"\n\n        if container is None:\n            raise ValueError(\"A container name must be provided.\")\n\n        # get Azure credentials\n        azure_credentials = Secret(azure_credentials_secret).get()\n\n        blob_service = azure.storage.blob.BlobServiceClient.from_connection_string(\n            conn_str=azure_credentials\n        )\n\n        # create key if not provided\n        if blob_name is None:\n            blob_name = str(uuid.uuid4())\n\n        client = blob_service.get_blob_client(container=container, blob=blob_name)\n\n        client.upload_blob(data, overwrite=overwrite)\n\n        return blob_name\n", "path": "src/prefect/tasks/azure/blobstorage.py"}]}