Azure function Not getting triggered or invocated.Getting 401 unauthorized on testing locally

Question

I am building an Azure Function using Python SDK with Cosmos DB Trigger to monitor the Change feed and write data to Blob storage. Additionally, I have a Stats.json blob that contains a key Stats with a value of 1 or 0, which I am updating as part of the process.

The development environment I'm using is Visual Studio Code, and I push changes to Azure Function from there. However, when I test the function locally, I encounter an error message: Unexpected status code: 401. Furthermore, the function is not triggering as expected in the Azure portal. Please note that I am not using any HTTP trigger in this scenario.

Any guidance or suggestions to resolve the "401 Unauthorized" error and get the Azure Function triggering successfully in the Azure portal would be greatly appreciated. Thank you!

Below is my __init__.py

import azure.functions as func
from azure.storage.blob import BlobServiceClient, BlobClient
import json
import logging
import os
import time

def get_recent_state_from_blob() -> int:
    try:
        blob_client=connect_storage()
        logging.info("Connected to blob storage")
        state_data = blob_client.download_blob().readall().decode("utf-8")
        state_json = json.loads(state_data)
        Stats = state_json.get("Stats", 0)
        return Stats
    except Exception as e:
        logging.error(f"Error while getting recent state from blob:{e}")
        return 0



def connect_storage() -> BlobClient:
    container_name='changedata'
    connection_string= os.environ['AzureWebJobsStorage']
    blob_name='Stats.json'
    try:
        blob_service_client = BlobServiceClient.from_connection_string(connection_string)
        container_client = blob_service_client.get_container_client(container_name)
        blob_client = container_client.get_blob_client(blob_name)
        return blob_client
    except Exception as e:
        raise e
    
def update_state_in_blob(new_state: int) -> None:
    try:
        blob_client=connect_storage()
        logging.info("Connected to blob storage")
        blob_data = blob_client.download_blob()
        blob_content = blob_data.readall()

        # Decode the existing JSON data
        data = json.loads(blob_content)

        # Update the 'State' key with value 1
        data['Stats'] = new_state

        # Add/Update the 'Time Updated' key with the current timestamp
        from datetime import datetime
        data['Time Updated'] = datetime.utcnow().isoformat()

        # Encode the data back to JSON format
        updated_content = json.dumps(data)

        # Upload the updated content to the blob
        blob_client.upload_blob(updated_content, overwrite=True)

        logging.info("Blob updated successfully.")
    except Exception as e:
        raise e

def main(
    documents: func.DocumentList, outputBlob: func.Out[str]
) -> None:
    consolidated_data = []

    if documents:
        for document in documents:
            logging.info("id: %s", document["id"])
            logging.info("SwitchNum: %s", document["SwitchNum"])
            logging.info("FileNum: %s", document["FileNum"])
            logging.info("CallingNum: %s", document["CallingNum"])
            consolidated_data.append(
                {
                    "id": document["id"],
                    "SwitchNum": document["SwitchNum"],
                    "FileNum": document["FileNum"],
                    "CallingNum": document["CallingNum"],
                }
            )

    data = {"consolidated_data": consolidated_data}
    json_data = json.dumps(data, indent=4)
    logging.info(json_data)
    outputBlob.set(json_data)

    state = get_recent_state_from_blob()

    if state == 2:
        update_state_in_blob(1)
        time.sleep(300)
        update_state_in_blob(0)
        logging.info("Record written successfully")
    else:
        logging.info("State is Active. Skipping...")

Function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "documents",
      "connectionStringSetting": "CosmosChangeFeedTrigger_ConnectionString",
      "databaseName": "UI_Trigger",
      "collectionName": "AF_Changefeed",
      "leaseCollectionName": "leases",
      "createLeaseCollectionIfNotExists": true,
      "direction": "in",
      "type": "cosmosDBTrigger"
    },
    {
      "connection": "AzureStorageConnection",
      "name": "outputBlob",
      "path": "changedata/files",
      "direction": "out",
      "type": "blob"
    }
  ]
}

My folder structure below:

Answer 1

I didn't run into a 404 issue but I did have an assembly load type issue. It was fixed when I updated my Azure Function Core Tools via winget upgrade Microsoft.Azure.FunctionsCoreTools to 4.0.5198. I'm also leveraging Azurite to emulate my storage when testing locally, allowing me to have "AzureWebJobsStorage": "UseDevelopmentStorage=true" in my local.settings.json. This resulted in no issues with local debugging. You may also want to look into new v2 programming model which uses decorators to configure bindings instead of a json file, making it more code centric.

Since you're reading and writing to blob, I leveraged this binding example to set both input and output strings. Code isn't exactly like yours, for instance I'm reading and writing directly to stats.json while you're writing document data to the blob, but basic principles are the same:

• trigger off cosmos change feed
• read from a blob
• copy change feed data to array
• write to the same blob

init.py

import json
import azure.functions as func

from datetime import datetime

def main(documents: func.DocumentList, inputblob:str, outputblob: func.Out[str]) -> None:
    movielist = []
    
    for document in documents:
        movielist.append(document.to_json())

    stats_json = {}
    if len(inputblob) > 0:
        stats_json = json.loads(inputblob)
    stats_json['num_movies'] = len(movielist)
    stats_json['time_updated'] = datetime.utcnow().isoformat()
    stats_data = json.dumps(stats_json, indent=4)
    outputblob.set(stats_data)

functions.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "cosmosDBTrigger",
      "name": "documents",
      "direction": "in",
      "leaseCollectionName": "leases",
      "connectionStringSetting": "AzureCosmosDb",
      "databaseName": "MovieList",
      "collectionName": "Items",
      "createLeaseCollectionIfNotExists": true
    },
    {
      "type": "blob",
      "name": "inputblob",
      "dataType": "string",
      "path": "changedata/stats.json",
      "connection": "AzureWebJobsStorage",
      "direction": "in"
    },
    {
      "type": "blob",
      "name": "outputblob",
      "dataType": "string",
      "path": "changedata/stats.json",
      "connection": "AzureWebJobsStorage",
      "direction": "out"
    }
  ]
}

requirements.txt

# Do not include azure-functions-worker in this file
# The Python Worker is managed by the Azure Functions platform
# Manually managing azure-functions-worker may cause unexpected issues

azure-functions
azure-storage-blob
requests
pandas

My local.settings.json has my Account Endpoint to my Cosmos. Since this value contains the Account Key, it's advised that you leverage something like App Config to store this value.

Once you're successfully running locally, then deploying your function via Visual Studio Code is a matter of making sure your storage account and CosmosDb endpoints are correctly configured under Application Settings on the Configuration blade.