prompt: 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()
  
  ```
  
golden_diff: 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
  
verification_info: {"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"}]}