[Python] Azure Function - Mange file transfer between blob containers

Dears,

Encountering an issue with an Azure function that manages file transfer between blob containers, the function is working locally by func command. However, once deployed on the cloud, it's not working with "Not Found" result testing. Check out the attached code samples.

function_app.py content

import logging
import azure.functions as func
from cp_blob import cp_blob
from csv_reader import csv_dest_path_handler, csv_reader, csv_src_path_handler

app = func.FunctionApp()
@app.blob_trigger(arg_name="myblob",
path="source",
connection="AzureWebJobsStorage" # the connection string placed into local
# setting file as well as function env vars
)
def data_connector_cp_blob(myblob: func.InputStream) -> None:
"""data_connector_cp_blob function."""

logging.info(f"Python blob trigger function processed blob"
f"Name: {myblob.name}"
f"Blob Size: {myblob.length} bytes")

# Extract blob name (strip off container prefix)
blob_name = myblob.name.split("/")[-1]

# Call the CSV reader function to get source and destination paths
paths_list = csv_reader()

# Call the CSV source paths
updated_src_paths = csv_src_path_handler(paths_list, blob_name)

# Call teh CSV destination paths
updated_dest_paths = csv_dest_path_handler(paths_list, blob_name)

# Call the cp_blob function to copy the blob
cp_blob(updated_src_paths, updated_dest_paths)




cp_blob.py content

"""Module to manage file transfer to/from Azure Blob Storage."""

import logging
from azure.storage.blob import BlobClient

def cp_blob(updated_src_paths, updated_dest_paths):
""" cp_blob function."""

for src_path, dest_path in zip(updated_src_paths, updated_dest_paths):
try:
logging.info(
"Source Path: %s, Destination Path: %s.",
src_path,
dest_path,)

dest_blob_client = BlobClient.from_blob_url(dest_path)

src_blob_client = BlobClient.from_blob_url(src_path)

if src_blob_client.exists():
copy_props = dest_blob_client.start_copy_from_url(src_path)
logging.info("copy started. copy ID %s.", copy_props['copy_id'])
continue

except Exception as e:
logging.error("An error occurred while copying the blob: %s", e)



Thank you

Continue reading...