An Azure service that provides an event-driven serverless compute platform.
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 screenshots attached and 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
Azure Functions
1 answer
Sort by: Newest
-
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