developing and deploying azure functions locally using VS code

Question

I'm trying to develop and deploy my http trigger locally from visual code studio for the past 2 days I'm able to do this but suddenly my http trigger function gone in azure portal and also not visible in VS code workspace when I try to start to debug. I'm getting following error.

Function code:

    import azure.functions as func
    from azure.storage.blob import BlobServiceClient
    from azure.iot.device import IoTHubDeviceClient, Message
    import crc32c
    import base64
     
    def main(req: func.HttpRequest) -> func.HttpResponse:
        # Retrieve the blob name from the HTTP request
        blob_name = req.params.get('blob_name')
        if not blob_name:
            return func.HttpResponse(
                "Please provide a 'blob_name' parameter in the query string.",
                status_code=400
            )
     
        # Connect to Azure Blob Storage
        blob_service_client = BlobServiceClient.from_connection_string("blob connection string")
        blob_client = blob_service_client.get_blob_client(container="blob-device", blob=blob_name)
     
        try:
            # Download the blob
            blob_data = blob_client.download_blob()
            binary_data = blob_data.readall()
     
            # Send the binary file to the device (pseudo-code)
            send_to_device(binary_data)
     
            return func.HttpResponse("Binary file sent to the device successfully.", status_code=200)
     
        except Exception as e:
            return func.HttpResponse(f"An error occurred: {str(e)}", status_code=500)
     
    def send_to_device(binary_data):
        # Calculate CRC32 checksum of the binary data
        crc32_checksum = crc32c.crc32(binary_data)
     
        # Convert binary data to base64 for transmission
        # encoded_data = base64.b64encode(binary_data)
     
        # Connect to IoT Hub
        conn_str = "Iot Hub connect string"
        device_client = IoTHubDeviceClient.create_from_connection_string(conn_str)
     
        try:
            # Connect to the IoT Hub
            device_client.connect()
     
            # Send the binary file along with CRC32 checksum as a message to IoT Hub
            msg = {
                "file": binary_data,
                "checksum": crc32_checksum
            }
            device_client.send_message(msg)
     
            print("Binary file sent to IoT Hub successfully.")
           
            # Receive acknowledgment or any other messages from IoT Hub
            while True:
                message = device_client.receive_message()
                if message:
                    print("Received message from IoT Hub:")
                    print(message.data)
                    # Handle the received message here
                    # Example: If the message contains acknowledgment or status
                    # handle_message(message)
                else:
                    print("No more messages from IoT Hub.")
                    break
           
        except Exception as e:
            print(f"An error occurred while sending data to IoT Hub: {e}")
        finally:
            # Disconnect from IoT Hub
            device_client.disconnect()

Update I'm able deploy the code and I copy the function URL. The following is the output I'm getting VS code and in browser

Updated code :

    import azure.functions as func
import logging
from azure.storage.blob import BlobServiceClient
from azure.iot.device import IoTHubDeviceClient, Message
import crc32c
import base64
import os
import json
import random
import sys
from azure.iot.hub import IoTHubRegistryManager
 
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    # Retrieve the blob name from the HTTP request
    blob_name = req.params.get('blob_name')
    if not blob_name:
        return func.HttpResponse(
            "Please provide a 'blob_name' parameter in the query string.",
            status_code=400
        )
 
    # Connect to Azure Blob Storage
    blob_service_client = BlobServiceClient.from_connection_string("DefaultEndpointsProtocol=https;AccountName=otastorage;AccountKey=5dCDeZPLUtEb5ucDZbJT8gosEB3P6LK92D4HRW244SnUU1xsgWDTfClJNVhLDonZ9cChAa8phO8+AStvBUT/w==;EndpointSuffix=core.windows.net")
    blob_client = blob_service_client.get_blob_client(container="blob-device", blob=blob_name)
 
    try:
        # Download the blob
        blob_data = blob_client.download_blob()
        binary_data = blob_data.readall()
 
        # Send the binary file to the device (pseudo-code)
        send_to_device(binary_data)
 
        return func.HttpResponse("Binary file sent to the device successfully.", status_code=200)
 
    except Exception as e:
        return func.HttpResponse(f"An error occurred: {str(e)}", status_code=500)
 
def send_to_device(binary_data):
    # Calculate CRC32 checksum of the binary data
    crc32_checksum = crc32c.crc32(binary_data)
 
    # Convert binary data to base64 for transmission
    # encoded_data = base64.b64encode(binary_data)
 
    # Connect to IoT Hub
    CONNECTION_STRING = "HostName=iothub.azure-devices.net;SharedAccessKeyName=service;SharedAccessKey=MmjcbQ0RCmuIK/JfGO28odyxCf9ZN7TL8AIoTHrJNvA="
    DEVICE_ID = "OTA-Device"
 
    try:
        # Create IoTHubRegistryManager
        registry_manager = IoTHubRegistryManager(CONNECTION_STRING)

        '''props={}
        # optional: assign system properties
        props.update(messageId = "message_%d" % i)
        props.update(correlationId = "correlation_%d" % i)
        props.update(contentType = "application/json")'''

        msg_body = {
            "file": base64.b64encode(binary_data).decode(),
            "checksum": crc32_checksum
        }
        msg = Message(json.dumps(msg_body))

        registry_manager.send_c2d_message(DEVICE_ID, msg)
    except Exception as ex:
        print ( "Unexpected error {0}" .format(ex) )
        return

The trigger is happening but the following error is appearing.

Invoke Screenshot

Updated Receive program python

import time
from azure.iot.device import IoTHubDeviceClient

RECEIVED_MESSAGES = 0

CONNECTION_STRING = "HostName=iothub.azure-devices.net;DeviceId=MCU;SharedAccessKey=pOkHxKaDejoJUSgUU3ZdmKqG/KV3XOcbgILSX8tDiMk="


def message_handler(message):
    global RECEIVED_MESSAGES
    RECEIVED_MESSAGES += 1
    print("")
    print("Message received:")

    # print data from both system and application (custom) properties
    for property in vars(message).items():
        print ("    {}".format(property))

    print("Total calls received: {}".format(RECEIVED_MESSAGES))
    
def main():
    print ("Starting the Python IoT Hub C2D Messaging device sample...")

    # Instantiate the client
    client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING, websockets=True)

    print ("Waiting for C2D messages, press Ctrl-C to exit")
    try:
        # Attach the handler to the client
        client.on_message_received = message_handler

        while True:
            time.sleep(1000)
    except KeyboardInterrupt:
        print("IoT Hub C2D Messaging device sample stopped")
    finally:
        # Graceful exit
        print("Shutting down IoT Hub Client")
        client.shutdown()
        
if __name__ == '__main__':
    main()

Answer 1

I have used the below code and it worked for me.

import azure.functions as func
import logging
from azure.storage.blob import BlobServiceClient
from azure.iot.device import IoTHubDeviceClient, Message
import crc32c
import base64
import os
import json

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    blob_name = req.params.get('blob_name')
    if not blob_name:
        return func.HttpResponse(
            "Please provide a 'blob_name' parameter in the query string.",
            status_code=400
        )
    else:
        # Connect to Azure Blob Storage
        blob_connect_str = os.environ["StorageAccountConnectionString"]
        blob_service_client = BlobServiceClient.from_connection_string(blob_connect_str)
        blob_client = blob_service_client.get_blob_client(container="demo-container", blob=blob_name)

        try:
            # Download the blob
            blob_data = blob_client.download_blob()
            binary_data = blob_data.readall()
            # Send the binary file to the device (pseudo-code)
            send_to_device(binary_data)

            return func.HttpResponse("Binary file sent to the device successfully.", status_code=200)

        except Exception as e:
            return func.HttpResponse(f"An error occurred: {str(e)}", status_code=500)

def send_to_device(binary_data):
    # Calculate CRC32 checksum of the binary data
    crc32_checksum = crc32c.crc32(binary_data)

    # Connect to IoT Hub
    iot_conn_str = os.environ["IotHubConnectionString"]
    device_client = IoTHubDeviceClient.create_from_connection_string(iot_conn_str)

    try:
        # Connect to the IoT Hub
        device_client.connect()

        # Send the binary file along with CRC32 checksum as a message to IoT Hub
        msg_body = {
            "file": base64.b64encode(binary_data).decode(),
            "checksum": crc32_checksum
        }
        msg = Message(json.dumps(msg_body))
        device_client.send_message(msg)

        print("Binary file sent to IoT Hub successfully.")

        # Receive acknowledgment or any other messages from IoT Hub
        while True:
            message = device_client.receive_message()
            if message is None:
                continue
            print("Received message from IoT Hub:")
            print(message.data)
            break

    except Exception as e:
        print(f"An error occurred while sending data to IoT Hub: {e}")
    finally:
            # Disconnect from IoT Hub
            device_client.disconnect() 

I have declared the connection strings in local settings file.

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "StorageAccountConnectionString": "DefaultEndpointsProtocol=https;AccountName=afrinstore01;AccountKey=cRWkgd************==;EndpointSuffix=core.windows.net",
    "IotHubConnectionString": "HostName={HostName}.azure-devices.net;DeviceId={deviceId};SharedAccessKey=3Rj*********k="
  }
}

requirements.txt-

azure-functions
azure-storage-blob
azure-iot-device
crc32c

I am able to send the message to Azure Iot Hub.

Below code used IoTHubRegistryManager to send a message from Storage account.

import azure.functions as func
import logging
from azure.storage.blob import BlobServiceClient
import crc32c
import base64
import os
import json
from azure.iot.hub import IoTHubRegistryManager
 
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    # Retrieve the blob name from the HTTP request
    blob_name = req.params.get('blob_name')
    if not blob_name:
        return func.HttpResponse(
            "Please provide a 'blob_name' parameter in the query string.",
            status_code=400
        )
    else:
        # Connect to Azure Blob Storage
        blob_connect_str = os.environ["StorageAccountConnectionString"]
        blob_service_client = BlobServiceClient.from_connection_string(blob_connect_str)
        blob_client = blob_service_client.get_blob_client(container="demo-container", blob=blob_name)
 
        try:
            # Download the blob
            blob_data = blob_client.download_blob()
            binary_data = blob_data.readall()
 
            # Send the binary file to the device (pseudo-code)
            send_to_device(binary_data)
 
            return func.HttpResponse("Binary file sent to the device successfully.", status_code=200)
 
        except Exception as e:
            return func.HttpResponse(f"An error occurred: {str(e)}", status_code=500)
 
def send_to_device(binary_data):
    # Calculate CRC32 checksum of the binary data
    crc32_checksum = crc32c.crc32(binary_data)
 
    # Connect to IoT Hub
    CONNECTION_STRING = "{IOT_connection_string}"
    DEVICE_ID = "{deviceId}"
    
    try:
        
        # Create IoTHubRegistryManager
        registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING)

        msg_body = {
            "file": base64.b64encode(binary_data).decode(),
            "checksum": crc32_checksum
        }
        msg = json.dumps(msg_body)
        registry_manager.send_c2d_message(DEVICE_ID, msg)
        
    except Exception as ex:
        print ( "Unexpected error {0}" .format(ex) )
        return