python-3.x 使用.csv作为API的输入

oo7oh9g9  于 2023-04-22  发布在  Python
关注(0)|答案(2)|浏览(118)

我有一个API,它使用post方法来更新数据库中的一些数据。这个API现在只允许每次调用一次更新,这意味着如果你想更新很多产品,你必须一个接一个地做。
示例:

payload = json.dumps({
  "supplierReference": "XX23", 
    "products" :[{
        "Distributor": "",
        "Description": "",
        "Price": ""
        "Brand": "",
    }    ]
}

我需要了解是否可能,如果可能的话,如何使用.csv文件或.xlsx文件作为API调用的输入,通过多次调用API(与行一样多的次数)来一次更新许多产品。
文件是这样的:

我的代码是这样的:

import pandas as pd
from pandas import json_normalize
import requests
import json

# Authentication part of the code

url = "https://api-gateway.com/api"

headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer '+ data['access_token'],
}

payload = json.dumps({
  "supplierReference": "XX23", 
    "products" :[{
  "Distributor": input("Enter Distributor: "), 
  "Description": input("Enter Description: "),
  "Price": input("Enter Price: "),
  "Brand": input("Enter Brand: ")
})

response = requests.post(url, headers=headers, data=payload, verify=True)

response_data = json.loads(response.text)
rsaldnfx

rsaldnfx1#

也许你正在寻找这样的东西。

with open(CSV_FILE, "r") as csv_f:
    next(csv_f)
    for line in csv_f:
        values = line.strip().split(",") #if ',' is the separator
        payload = {
            "Reference": values[0],
            "Distributor": values[1],
            "Description": values[2],
            "Price": values[3],
            "Brand": values[4]
        }
        response = requests.post(url, headers=headers, data=payload, verify=True)
        response_data = json.loads(response.text)
qmb5sa22

qmb5sa222#

验证您正在使用的API是否不支持JSON列表。如果不支持,您将不得不在循环中执行此操作,每个项目都创建一个新请求。或者您可以使用aiohtp使用异步请求:

import asyncio
import aiohttp

url = "https://api-gateway.com/api"
my_data = ['{...}', '{...}', '{...}'] # imagine an Iterable containing your JSON data

async def send_post_request(session, url, data):
    async with session.post(url, data=data) as response:
        response_text = await response.text()
        print(f"Response from {url}: {response_text}")

async def main():
    async with aiohttp.ClientSession() as session:
    tasks = [asyncio.create_task(send_post_request(session, url, item)) for item in my_data]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

如果API本身可以并行处理请求,那么像这样异步执行很可能会使它非常快。如果同时发送太多请求,可能会超过API的速率限制。

相关问题