使用 Python 和 Azure 函数将 URL 参数保存为 CSV 文件

Question

我想用这样的 HTTP POST 发送一些数据：

https://httptrigger-testfunction.azurewebsites.net/api/HttpTrigger1?id
=test&serial_id
=1254&device_tra
=302&received_time
=2021-03-01

我从here 的 Microsoft 示例中编写了一个 Azure 函数，该函数从 HTTP POST 读取“名称”。 现在，我想读取上述数据并将其保存到 Blob 存储上的 CSV 文件中。 我应该使用哪个模块？

示例代码：

import logging

import azure.functions as func


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

    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
        )

Answer 1

请参考我的代码：

import logging
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
import azure.functions as func
import os, uuid
import tempfile


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

    connect_str = "<your-connection-string>"
    container_name = "<your-container-name>"

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

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

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

    # Create the BlobServiceClient object which will be used to create a container client
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)

    # Create the container
    container_client = blob_service_client.get_container_client(container_name)

    # Create a local directory to hold blob data
    local_path = tempfile.gettempdir()

    # Create a file in the local data directory to upload and download
    local_file_name = str(uuid.uuid4()) + ".csv"
    upload_file_path = os.path.join(local_path, local_file_name)
    logging.info(upload_file_path)

    # Write text to the file
    file = open(upload_file_path, 'w')
    csv_content = id + "," + serial_id + "," + device_tra + "," + received_time
    logging.info(csv_content)
    
    file.write(csv_content)
    file.close()

    # Create a blob client using the local file name as the name for the blob
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=local_file_name)

    print("\nUploading to Azure Storage as blob:\n\t" + local_file_name)

    # Upload the created file
    with open(upload_file_path, "rb") as data:
        blob_client.upload_blob(data)

    if id:
        return func.HttpResponse(f"Hello, {id}. 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
        )

或者你可以使用这个代码：

import logging
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
import azure.functions as func
import os, uuid
import tempfile
import csv


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

    connect_str = "<your-connection-string>"
    container_name = "<your-container-name>"

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

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

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

    # Create the BlobServiceClient object which will be used to create a container client
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)

    # Create the container
    container_client = blob_service_client.get_container_client(container_name)

    # Create a local directory to hold blob data
    local_path = tempfile.gettempdir()

    # Create a file in the local data directory to upload and download
    local_file_name = str(uuid.uuid4()) + ".csv"
    upload_file_path = os.path.join(local_path, local_file_name)
    logging.info(upload_file_path)

    with open(upload_file_path, 'w', newline='') as csvfile:
        filewriter = csv.writer(csvfile, delimiter=',',
                                quotechar='|', quoting=csv.QUOTE_MINIMAL)
        filewriter.writerow(['id', 'serial_id', 'device_tra', 'received_time'])
        filewriter.writerow([id, serial_id, device_tra, received_time])

    # Create a blob client using the local file name as the name for the blob
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=local_file_name)

    print("\nUploading to Azure Storage as blob:\n\t" + local_file_name)

    # Upload the created file
    with open(upload_file_path, "rb") as data:
        blob_client.upload_blob(data)

    if id:
        return func.HttpResponse(f"Hello, {id}. 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
        )

Answer 2

您可以使用storage account Python SDK。这个适用于 ADLS Gen 2，如果您使用 Gen 1 或更早版本，请找到正确的SDK here。

查看“uploading a file”部分，它展示了如何将字符串数据写入 blob。例如。在下面的代码中，您希望将 csv 内容放入变量 data。

from azure.storage.filedatalake import DataLakeFileClient

data = b"abc"
file = DataLakeFileClient.from_connection_string("my_connection_string",
                                                 file_system_name="myfilesystem", file_path="myfile")

file.append_data(data, offset=0, length=len(data))
file.flush_data(len(data))

还有一些samples here。

您需要使用一些 csv writer 写入内存中的字符串（可能是 StringIO）而不是本地文件，然后将该字符串写入 ADLS。

---

如果从代码和示例中看不出来，您可以使用req.params 来访问参数。除了 body 没有任何东西有 getter。