使用 Python 将 Azure Blob 存储数据直接推送到 POST 请求中

Question

我正在努力将数据从 Azure Blob 发送到接受 POST 请求以上传应为 xlsx 或 csv 格式的文件的平台。

为了使我的代码正常工作，首先需要使用 open() 库将 blob 数据写入文件，然后将文件作为 io.BufferedReader 读回算法中 然后可以在 requests.post 文件参数中使用它，这是一个 json

问题： 我的问题是我需要使用 azure function app 在生产环境中运行此代码。应用程序不允许我简单地使用 os 和 open() 我尝试使用 io.BytesIO 从 blob 获取字节数据，但 API 无法将其识别为 csv 或 xlsx 文件

有没有办法使用内存中的数据并将其用于 POST 请求？

这是我的代码：

# GETTING BLOB DATA AND UPLOADING IT USING POST REQUEST

from datetime import datetime, timezone, timedelta
import requests
import json
import os
from requests.auth import HTTPBasicAuth
from azure.storage.blob import BlobClient

# Get excel or csv file from azure blob storage (stored as blob)

# 1. Getting the blob data
connect_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING") # Hard coding key when in production
container_name = 'CONTAINER_NAME'
blob_name = 'TEST.xlsx'
blob_client = BlobClient.from_connection_string(connect_str,container_name,blob_name)
data = blob_client.download_blob()

# Writing the Data into a file and then
# Reading back into the program to create
# io.BufferedReader <name='TEST.xlsx'> object for POST request

with open(blob_name,'wb') as file_content:
    file_content.write(data.readall())
    file_content.close()

f_data = open(blob_name,'rb')

_id = 'jkdfhiuweh93o0' # list_id for post request
file_name = "TEST"
url = "https://platform.com/api/uploadFile" # API endpoint
auth = {"X-Gatekeeper-SessionToken": "915760344f3a450e83e5fae98a78d87e"}
body = {"id": _id,'listname':file_name}

# Making a POST request and uploading the data
file_upload = requests.post(url, headers=auth, files=f_data, data=body)
print(file_upload.json())

我一直在寻找解决方案好几个星期，但没有运气...... 非常感谢任何 cmets 和帮助。

Answer 1

最后，使用requests.post并嵌入一个不是从网上获取的文件的方法是使用tempfile库。

解决办法

import tempfile, requests, json
from azure.storage.blob import BlobClient

token = {"X-Gatekeeper-SessionToken": "2444e9b6e3544cae6d8e9e6f4a5894"}
connect_str = '<Connection String - App Key 1>' # Obtained from azure storage account
container_name = '<Container Name>'
blob_name = "TEST.xlsx"
blob_client = BlobClient.from_connection_string(connect_str,container_name,blob_name)
data = blob_client.download_blob()

file_name = blob_name.split(".")[0]
file_format = blob_name.split(".")[-1]

# Temporarily Storing file in temp directory
temp_file = tempfile.NamedTemporaryFile('w+b', prefix=file_name, suffix="."+file_format)
temp_file.write(data.readall())
access_file = open(temp_file.name,'rb')

f_data = {'file':access_file,'Name':blob_name}

_id = 12345 #Each list (TEST.xlsx) has an id associated with it
if _id == 404:
    body = {"listname":file_name}
else:
    body = {"id": _id,'listname':file_name}

file_name = blob_name.split(".")[0]

url = "https://manage.fastfieldforms.com/api/lookuplist"


resp = requests.post(url, headers=token, files=f_data, data=body)
print(resp.json())
access_file.close()
temp_file.close()