提问人:Jose Moquiambo 提问时间:6/15/2023 最后编辑:Jose Moquiambo 更新时间:6/15/2023 访问量:140
为什么我的 API 响应总是无效?我的call_api功能有问题吗?[已解决]
Why are my API response always invalid ? Is there something wrong with my call_api function?[Solved]
问:
在我的call_api函数中,我使用了一种不同的方法来检测无效/有效的 API,即检查标头“Response_Description”中是否存在字符串“Hazard 已成功处理并返回至少一个值”。不知何故,我的输出总是 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
答:
1赞
CtrlZ
6/15/2023
#1
重构 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
评论
0赞
Jose Moquiambo
6/15/2023
不幸的是,这似乎不起作用,因为我仍然为我的所有 API 获得 False。如果所有(结果)其他“N”,则此行是否有问题,传递 = “Y”。
评论