json 为什么我的API响应总是无效?我的call_API函数有问题吗?[已解决]

44u64gxh  于 2023-06-25  发布在  其他
关注(0)|答案(1)|浏览(120)

在我的call_API函数中,我使用了一种不同的方法来检测无效/有效的API,即检查字符串“Hazard was successfully processed and the至少一个value is return.”是否存在于标题“Response_Description”中。不知何故,我的输出总是为false,在Pass列下的输出表中标记为N。

def call_api(url):
try:
    headers = {
        "User-Agent": "Mozilla/5.0 (X11; CrOS x86_64 12871.102.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.141 Safari/537.36"
    }
    response = requests.get(url, headers=headers)
    json_response = response.json()
    if (
        "Response_Description" in json_response
        and "Hazard was successfully processed" in json_response["Response_Description"]
    ):
        return True
    else:
        return False
except requests.exceptions.RequestException:
    return False

df = pd.read_csv(r"C:\Users\Jose.Moquaimbo\Bulk Calling APIs\dataset.csv")

# Number of iterations
num_iterations = 5

# Create an empty DataFrame to store the results
results_df = pd.DataFrame(columns=["Iteration", "Pass", "Time Taken"])

# Variables for tracking min, max, and total time
min_time = float("inf")
max_time = float("-inf")
total_time = 0

def process_iteration(iteration):
# Get a random sample of URLs from the DataFrame
 random_urls = df["url"].sample(n=1).tolist()

# Execute API calls concurrently using ThreadPoolExecutor
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    start_time = time.time()
    futures = [executor.submit(call_api, url) for url in random_urls]

# Wait for all futures to complete and get their results
results = [future.result() for future in futures]

# Stop timer
end_time = time.time()

# Calculate the time taken for this iteration
iteration_time = end_time - start_time

# Update min, max, and total time
global min_time, max_time, total_time
min_time = min(min_time, iteration_time)
max_time = max(max_time, iteration_time)
total_time += iteration_time

# Check if any API call was not successful in this iteration
passed = "Y" if all(results) else "N"

# Add the iteration results to the DataFrame
results_df.loc[iteration] = [iteration, passed, iteration_time]

# Run the iterations
for i in range(1, num_iterations + 1):
    process_iteration(i)

# Calculate average time per iteration
avg_time = total_time / num_iterations

# Display the results DataFrame
print(results_df)

# Summary statistics
print("Minimum time taken:", min_time)
print("Maximum time taken:", max_time)
print("Average time per iteration:", avg_time)
print("Y stands for error-free response and N for invalid response")

输出

Iteration Pass  Time Taken
 1          1    N    0.276398
 2          2    N    0.298180
 3          3    N    0.307337
 4          4    N    0.323730
 5          5    N    0.333215
 Minimum time taken: 0.2763981819152832
 Maximum time taken: 0.33321452140808105
 Average time per iteration: 0.3077719688415527
 Y stands for error-free response and N for invalid response

有没有一种方法可以修改代码,这样它就可以正确地验证API,因为似乎API总是无效的。我怀疑我的call_API函数有问题
编辑:

response_text = response.text
wb1gzix0

wb1gzix01#

重新构造call_API()函数以更好地处理异常并报告可能出现的任何问题。

import requests
import sys

HEADERS = {
    'User-Agent': 'Mozilla/5.0 (X11; CrOS x86_64 12871.102.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/81.0.4044.141 Safari/537.36'
}
KEY = 'Response_Description'
VALUE = 'Hazard was successfully processed'

def call_api(url: str) -> bool:
    try:
        with requests.get(url, headers=HEADERS) as response:
            response.raise_for_status()
            return VALUE in response.json().get(KEY, '')
    except Exception as e:
        print(e, file=sys.stderr)
    return False

相关问题