Event Subscription Storage Blob create/delete not triggered via azure cli, azcopy or python SDK

Hello @Sina Salam Thank you for your response.

There are no error logs to share as no errors are occurring, just a lack of events. Below is the python code I have been using to upload and delete blobs, for testing. I have also run the cli commands manually with the same outcome.

The storage account is Gen2 storage. If you wish for further info on this please tell me which CLI command I should use.

I made my Event Subscription though the portal, and in my filters I have set subject starts with /blobServices/default/containers/container_name/blobs/directory_name and have all 10 available event types selected "Blob Created, Blob Deleted, Directory Created, Directory Deleted, Blob Renamed, Directory Renamed, Blob Tier Changed, Blob Inventory Completed, Async Operation Initiated, Lifecycle Policy Completed" but when running the below code no events are triggered other than actions through the portal, I also do not see any messages in the service bus queue nor updates in the event subscription metrics graph.

How can I check the permissions of the identities with regards to triggering event subscriptions?

I have also now looked into the storage account activity logs. I can see the Event subscription attaching itself whenever I update it but uploading the code via the SDK or CLI illicit no logged activity.

I will also retry using the REST API in the mean time.

Thank you for your help.

Python code to upload and delete a blob to Azure blob store. Login to azure cli before running code.

# %% --------------------------------------------------------------------------
# Imports and Constants
# -----------------------------------------------------------------------------
#region Imports_and_Constants
# Standard Library
import os
from pathlib import Path
from psutil._common import bytes2human
from datetime import datetime, timedelta, timezone
import logging
from pprint import PrettyPrinter
import subprocess
from subprocess import CompletedProcess, CalledProcessError
from typing import Dict
# 3rd Party
from azure.identity import DefaultAzureCredential, AzureCliCredential
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, BlobPrefix, BlobProperties
from azure.servicebus import ServiceBusClient, ServiceBusReceivedMessage, ServiceBusMessage, ServiceBusSender
import azure.core.exceptions as Azexp

# Custom



# Constants & Globals
NO_PROXY = { "http":"", "https":"" }

pp = PrettyPrinter(indent=1)
logging.basicConfig(level="DEBUG")
log = logging.getLogger(__name__)
# Notes


# %% --------------------------------------------------------------------------
# Azure utils
# -----------------------------------------------------------------------------
#region Azure Utils
def get_BlobServiceClient(account_url:str, credentials:str|Dict[str, str]|None=None, proxies:Dict[str,str]=None )->BlobServiceClient:
    """Get standard blob service client for using this library.
    Can use None/DefaultAzureCredential if azure cli has already been logged in. \"$ az login\" on runners \"$ az login --identity\"
    
    src: https://zcusa.951200.xyz/en-us/python/api/azure-storage-blob/azure.storage.blob.blobserviceclient?view=azure-python
    proxy docs: https://zcusa.951200.xyz/en-us/azure/developer/python/sdk/azure-sdk-configure-proxy?tabs=cmd#custom-configuration
    
    Args:
        account_url (str): Storage account URL, e.g. \"https://<storage-account-name>.blob.core.windows.net\"
        credentials (str | Dict[str, str] | None, optional): Credentails, If already logged in using the azure cli, then give None and DefaultAzureCredential will be used. Defaults to None.
        proxies (Dict[str,str] | None, optional): A dictionary with the keys, \"http\" and/or \"https\" and their respective proxy URLS. If not given will use default/environment proxy settings. If usernames or passwords may contain special characters use urllib.parse.quote() to encode them properly. 
    Returns:
        BlobServiceClient: BlobServiceClient object.
    """
    if not credentials:
        credentials = DefaultAzureCredential() # Assumes terminal/client is already logged in
        
    client = BlobServiceClient(
        account_url=account_url,
        credential=credentials,
        proxies=proxies)
    return client

def blob_exists(client_blobsvc: BlobServiceClient, container_name:str, pathblob:str)->bool:
    """Return if a blob exists.

    Args:
        client_blobsvc (BlobServiceClient): Blob service client
        container_name (str): Container name e.g. \"container1\"
        pathblob (str): Path in container to blob.

    Returns:
        bool: True if blob exists.
    """
    client_blob = client_blobsvc.get_blob_client(container=container_name, blob=pathblob)
    output = client_blob.exists()
    return output


def blob_upload_from_file(client_blobsvc:BlobServiceClient, container_name:str, pathfile_src:Path|str, pathblob_dest:Path|str, overwrite=True)->None:
    """Upload a blob to Azure blob store.
    Add \"/\" to pathfile_dest to indicate parent directories. Parent directorise will automatically be created.

    src: https://zcusa.951200.xyz/en-us/azure/storage/blobs/storage-blob-upload-python
    src2: https://github.com/Azure-Samples/AzureStorageSnippets/blob/master/blobs/howto/python/blob-devguide-py/blob-devguide-upload.py
    
    Args:
        client_blobsvc (BlobServiceClient): Blob service client
        container_name (str): Container name e.g. \"container1\"
        pathfile_src (Path|str): File path to file to upload.
        pathfile_dest (Path|str): Path representing where to upload the file to excluding the container. Add \"/\" delimiteres to place in directory, parent fdirectories will be automatically created. e.g. \"dir0/dir1/filename.txt\"
        overwrite (bool, optional): Whether to overwrite the file if it exsits. Defaults to True.
    Raises:
        azure.core.exceptions.ResourceExistsError, ErrorCode:BlobAlreadyExists: When overwrite is False and blob already exists 
    """
    # Can add / in name to specify subdirectory
    client_container = client_blobsvc.get_container_client(container=container_name)
    file_size = bytes2human(os.path.getsize(pathfile_src))
    log.debug("Uploading file: {}, {} -> {}/{}".format(file_size, str(pathfile_src), container_name, str(pathblob_dest)))
    dt_start = datetime.now()
    with open(pathfile_src, mode="rb") as f:
        client_blob = client_container.upload_blob(name=str(pathblob_dest), data=f, overwrite=overwrite)
    log.debug("Upload complete: {}, {} -> {}/{}".format(file_size,(datetime.now()-dt_start),container_name,str(pathblob_dest)))
    return None

def blob_upload_from_file_cli(storage_account_name:str, container_name:str, pathfile_src:Path, pathblob_dest:str )->CompletedProcess:
    """Upload file using az cli commands.
    Assumes az cli is already logged in. Run `az login` in terminal before running this code`
    Args:
        storage_account_name (str): Name of Azure blob storage account.
        container_name (str): Name of container within sotrage account.
        pathfile_src (Path): Path to file to upload.
        pathblob_dest (str): Path to location file should be uploaded to.

    Returns:
        CompletedProcess: Completed process of subroutine, with output captured.
    """
    log.debug("AZ cli uploading: {} -> {}/{}".format(str(pathfile_src), container_name, str(pathblob_dest)))
    dt_start = datetime.now()
    cmd = "no_proxy=\"*\" ;NO_PROXY=\"*\" ;az storage blob upload --account-name \"{}\" --container-name \"{}\" --auth-mode login --overwrite --file \"{}\" --name \"{}\"".format(storage_account_name, container_name, str(pathfile_src), str(pathblob_dest))
    cp = subprocess.run(cmd, shell=True, capture_output=True)
    if "ERROR" in cp.stderr.decode("utf-8"):
        log.warning("File upload may have failed Check following outputs.\nstdout:\n{}\nstderr:\n{}".format(cp.stdout.decode("utf-8"),cp.stderr.decode("utf-8")))
    else:
        log.debug("AZ cli upload complete - {} file:{} \nstdout:\n{}\nstderr:\n{}".format(
            datetime.now()-dt_start, str(pathfile_src), cp.stdout.decode("utf-8"), cp.stderr.decode("utf-8")))
    return cp

def blob_delete(client_blobsvc: BlobServiceClient, container_name:str, pathblob:str):
    """Delete given blob.
    Blob must exists, and if directory blob then must be empty.
    Use blob_exists() function to check if blob exists first.
    Args:
        client_blobsvc (BlobServiceClient): 
        container_name (str): Container name e.g. \"container1\"
        pathblob (str): Path to blob/full blob name e.g. \"blob.txt\" or \"dir0/blob.txt\"
        
    Raises:
        azure.core.exceptions.ResourceNotFoundError, ErrorCode:BlobNotFound: When given blob file does not exist in given container e.g. \"fakefile.txt\".
        azure.core.exceptions.HttpResponseError, ErrorCode:InvalidUri: When given path does not exist in given container e.g. \"fakedir/\".
        azure.core.exceptions.ResourceExistsError, ErrorCode:DirectoryIsNotEmpty: When directory blob is not empty.
    """
    log.debug("Deleting blob: {}/{}".format(container_name, pathblob))
    client_blob = client_blobsvc.get_blob_client(container=container_name, blob=pathblob)
    client_blob.delete_blob()

#endregion 


def main():
    storage_account_name = "storage_account_name"
    az_storage_url = "https://{}.blob.core.windows.net".format(storage_account_name)
    container_name = "container_name"
    pathblobdir_upload_test = "/dir/upload_test.txt"
    client_blobsvc = get_BlobServiceClient(az_storage_url, AzureCliCredential(), NO_PROXY)
    
    # Make test file
    pathfile_test = Path(__file__).parent/"upload_temp.txt"
    pathfile_test.parent.mkdir(parents=True, exist_ok=True)
    with open(pathfile_test,"w",encoding="utf-8") as f:
        f.write("This is a test upload file")
    # Step though code an watch event subscription/service bus queue for events.
    # Test if "Microsoft.Storage.BlobCreated" event is triggered on each upload.
    
    # Delete test file if it exists before testing
    if blob_exists(client_blobsvc, container_name, pathblobdir_upload_test):
        blob_delete(client_blobsvc, container_name, pathblobdir_upload_test)
    # Upload file to blob store using Python SDk
    blob_upload_from_file(client_blobsvc, container_name, pathfile_test, pathblobdir_upload_test, True)
    blob_delete(client_blobsvc, container_name, pathblobdir_upload_test)
    # Upload via cli
    blob_upload_from_file_cli(storage_account_name, container_name, pathfile_test, pathblobdir_upload_test)
    blob_delete(client_blobsvc, container_name, pathblobdir_upload_test)
    
    pathfile_test.unlink()
    return None




if __name__ == "__main__":
    main()
else:
    pass

Hello @Sina Salam

I accidentally posted my response to your comment as it own comment instead of as response to your comment.

Many Thanks

Barnaby

Hello @Sina Salam

I tired to respond to your answer and provide my code as requested, but my comment was deleted for a violation of the code of conduct. I will rewrite my answer and hopefully it is not deleted again.

Hello Ayriss, Barnaby (The Bridge),

Thank you for your feedback.

Yes! I saw deleted one, and I have restored it. After careful examining the content, since there are no errors, focusing on permissions, diagnostic logging, and verifying Event Grid filters and metrics should help pinpoint the issue.

Success.