How to output to Blob service with Durable functions

Question

I am new to Azure functons and Durable functions. I want to write a code that the activity function called by the orchestrator function will compute some value and connect to the Blob service.

How can I modify the code below to output the value generated by the activity function to the Blob service?

The environment I am using is,

Azure Functions, Durable Functions
Visual Studio Code
Python programming model v2

The following is a simplified code I am considering using. Please modify this one. Incidentally, client functions and orchestration functions may not have much to do with each other

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

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) -> dict:
    test  =  yield  context.call_activity("main", "")
    return {"Test": test} 

### 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]):
    string  =  "Data is successfully Inserted"
    logging.info(f'Python Queue trigger function processed {len(string)} bytes')
    outputblob.set(string)
    return  "Completed"

local_setting.json

{
  "IsEncrypted": false,
  "Values": {
  "AzureWebJobsStorage": "UseDevelopmentStorage=true",
  "FUNCTIONS_WORKER_RUNTIME": "python",
  "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
  "BlobStorageConnection": "DefaultEndpointsProtocol=https;AccountNa***"
  }
}

error message enter image description here

Ikhtesam Afrin · Accepted Answer · 2023-10-20 07:57:34Z

0

I have used the below code to write the value generated by activity function into blob storage.

function_app.py

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

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) -> dict:
test  =  yield  context.call_activity("main", "")
return {"Test": test} 

### activity function ###
# I want to calculate the value and output it to the bound service.
@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]):
string  =  "Data is successfully Inserted"
logging.info(f'Python Queue trigger function processed {len(string)} bytes')
outputblob.set(string)
return  "Completed"

local.settings.json

{
"IsEncrypted": false,
"Values": {
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
"BlobStorageConnection": "DefaultEndpointsProtocol=**********"
}
}

Fetch the value of connection string as shown below and add it in local.settings -

enter image description here

Output:

enter image description here

answered Oct 20, 2023 at 7:57

Ikhtesam Afrin

6,6902 gold badges4 silver badges11 bronze badges

Sign up to request clarification or add additional context in comments.

21 Comments

TY00 Over a year ago

Thank you. [connection="BlobStorageConnection"] in the activity function and in local.settings.json ["BlobStorageConnection": "DefaultEndpointsProtocol=**********"] are set to the same value?

Ikhtesam Afrin Over a year ago

@TY00 Yes, I am passing the value of BlobStorageConnection in "BlobStorageConnection": "DefaultEndpointsProtocol=**********"

TY00 Over a year ago

I copied and pasted all of function_app.py and local.settings.json and changed only one place in the connection string. I also set up requirements.txt. However, [Orchestrator function 'orchestrator' failed: Activity function 'main' failed: Storage account connection string 'AzureWeb Make sure that it is a defined App Setting. What is the cause?

Ikhtesam Afrin Over a year ago

@TY00 Are you passing the connection string in local.settings file and referring it in function_app.py ?

TY00 Over a year ago

All done without incident! Your help is much appreciated! Thank you so much!

|

Collectives™ on Stack Overflow

How to output to Blob service with Durable functions

1 Answer 1

21 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

21 Comments

Your Answer

Sign up or log in

Post as a guest

Related