How to use Celery to connect to Azure Blob Storage for background callback

Dash Python
1

Hi everyone, I followed the instruction from Background Callbacks | Dash for Python Documentation | Plotly and some celery documents to generate a connection to run the example but was unsuccessful.

I have access to a Azure Storage account. Tried a few ways to generate SAS key, etc. for broker parameter, but nothing seems to work. I’m not very experienced with this ,I would be glad if someone can help me. I have included a sample code below using the sample from Dash website.

import time
import os

from dash import Dash,CeleryManager, Input, Output, html, callback
from celery import Celery

access_key = os.environ['blob_accountkey']
account_name = os.environ['blob_accountname']

#format:     azurestoragequeues://:STORAGE_ACCOUNT_ACCESS kEY@STORAGE_ACCOUNT_NAME
#(https://docs.celeryq.dev/projects/kombu/en/v5.2.3/reference/kombu.transport.azurestoragequeues.html)
broker = f'azurestoragequeues://:{access_key}:{account_name}'

#The required URL format is azureblockblob:// followed by the storage connection string. 
# You can find the storage connection string in the Access Keys pane of your storage account resource in the Azure Portal
#(https://docs.celeryq.dev/en/stable/userguide/configuration.html#conf-azureblockblob-result-backend)

cache_dir = 'cache_dir'
backend = f'azureblockblob://DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={access_key};EndpointSuffix=core.windows.net/{cache_dir}'

celery_app = Celery(__name__, broker=broker, backend=backend)
background_callback_manager = CeleryManager(celery_app)



app = Dash(__name__)

app.layout = html.Div(
    [
        html.Div([html.P(id="paragraph_id", children=["Button not clicked"])]),
        html.Button(id="button_id", children="Run Job!"),
    ]
)

@callback(
    output=Output("paragraph_id", "children"),
    inputs=Input("button_id", "n_clicks"),
    background=True,
    manager=background_callback_manager,
)
def update_clicks(n_clicks):
    time.sleep(2.0)
    return [f"Clicked {n_clicks} times"]


if __name__ == "__main__":
    app.run(debug=True)