How to bind Durable functions and append them to text

Question

In Azure Durable functions, I want to bind to Blob storage and write to a text file.

The following code will write it as a new file instead of writing it as an additional file.

What is the code to write as an addition to text.txt in newblob?

import  azure.functions  as  func
import  azure.durable_functions  as  df
import  logging  
import  json

app  =  df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)  

### client function ###
@app.route(route="orchestrators/client_function")
@app.durable_client_input(client_name="client")
async  def  client_function(req: func.HttpRequest, client: df.DurableOrchestrationClient) -> func.HttpResponse:
    instance_id  =  await  client.start_new("orchestrator", None, {})
    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    await  client.wait_for_completion_or_create_check_status_response(req, instance_id)
    status  =  await  client.get_status(instance_id)
    return  f"output: {status.output}"  

### orchestrator function ###
@app.orchestration_trigger(context_name="context")
def orchestrator(context: df.DurableOrchestrationContext) -> str:
    result = yield context.call_activity("main", '')
    return "Inserted"

### activity function ###
@app.blob_output(arg_name="outputblob", path="newblob/test.txt", connection="BlobStorageConnection")
@app.activity_trigger(input_name="blank")
def main(blank: str, outputblob: func.Out[str]):
    data = {"size": 100, "time": 0.1}
    json_data = json.dumps(data)
    outputblob.set(json_data)
    return "Inserted"

Answer 1

In Azure Functions, when you bind to Blob Storage to write data, the behavior is to create or overwrite the specified blob. If you want to append to an existing blob, you have to handle that logic yourself and not use binding

To append data to a blob in Azure Functions:

1. Use the Azure Blob Storage SDK.
2. Download the existing content of the blob.
3. Append new data to it.
4. Re-upload the combined data.

Modified main function to achieve this as an example:

from azure.storage.blob import BlobServiceClient

### activity function ###
@app.activity_trigger(input_name="blank")
def main(blank: str):
    data = {"size": 100, "time": 0.1}
    json_data = json.dumps(data)
    
    # Connect to Blob Storage
    connection_str = "YOUR_CONNECTION_STRING"
    blob_service_client = BlobServiceClient.from_connection_string(connection_str)
    blob_client = blob_service_client.get_blob_client(container="newblob", blob="test.txt")

    # Download, append, and re-upload
    blob_data = blob_client.download_blob().readall() + json_data.encode()
    blob_client.upload_blob(blob_data, overwrite=True)

    return "Inserted"

Replace "YOUR_CONNECTION_STRING" with your Blob Storage connection string.