python - 使用 Azure Functions 中的 Python 根据 Azure Blob 存储中的模式匹配检查 Blob 是否存在

Question

有没有办法使用 Azure Function 基于模式使用 Python 检查文件是否存在于 Azure BLOB 容器中？我传入了不同的文件，我需要根据“FileName”参数来处理它们。放置在容器上的文件将带有附加日期。我需要检查该文件是否存在，然后对其进行处理。

容器中的示例文件=>

1. API_File1_20202008.CSV

2. API_File2_20202008.CSV

3. API_File3_20202008.CSV

如果 Function 中传入参数 = > API_File1。
然后该函数应检查 API_File1* 指定的任何 blob 是否存在，然后进行处理。

对于本地操作系统，我可以使用以下内容。

for name in glob.glob(SourceFolder +'/' + FileName + '*.CSV'):
    print(name)

有什么建议吗？

Answer 1

对于这个需求，您可以使用下面的代码在azure上实现。

import logging

import azure.functions as func
from azure.storage.blob.blockblobservice import BlockBlobService


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    #just focus on below part
    block_blob_service = BlockBlobService(account_name="<your storage name>", account_key="<storage key>")

    if block_blob_service.exists("<container name>", "file5.csv"):
            print("========found file=======");
    else:
            print("========not exist========");
    #just focus on the part above

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
    else:
        return func.HttpResponse(
             "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

在运行代码之前，您需要安装模块。 不要安装pip install azure-storage-blob，我使用安装azure-storage-blob进行测试，但出现了一些问题。您可以在“终端”窗口中的虚拟环境中安装pip install azure==4.0.0(可能需要一些时间)，然后代码就可以成功运行。

======================================更新==== ===============================

我认为我们不能直接通过"File5"+ "*.CSV"查询文件名。但我可以提供一个解决方法供你引用。然后函数获取请求参数fileName然后使用下面的代码:

files = [blob for blob in block_blob_service.list_blobs("testcontainer") if blob.name.startswith(fileName)]

for file in files:
    logging.info(file.name)

该代码用于获取以发送给函数的 fileName 开头的所有文件。