Azure Function - Mange file transfer between blob containers

Mohamed Eraki 0 Reputation points
2025-05-22T16:02:01.7633333+00:00

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 screenshots attached and code samples.

Screenshot_22-5-2025_185443_portal.azure.com

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

Azure Functions
Azure Functions

An Azure service that provides an event-driven serverless compute platform.


1 answer

Sort by: Newest
  1. Deleted

    This answer has been deleted due to a violation of our Code of Conduct. The answer was manually reported or identified through automated detection before action was taken. Please refer to our Code of Conduct for more information.

    5 deleted comments

    Comments have been turned off. Learn more

Your answer

Answers can be marked as 'Accepted' by the question author and 'Recommended' by moderators, which helps users know the answer solved the author's problem.