[Python] API Throttling Help Requested

Hi I am having trouble with API throttling causing 429 status errors

I had been using a version of this code in the past and thought I was pulling all data needed, and recently realised that I was missing a lot of data due to throttling

Was wondering if anyone had advice in this

The api I am using is from the OpenFDA: https://open.fda.gov/apis/device/event/

The documentation says there is a limit of 240 requests per minute but that would make this code too slow

import os, requests, aiohttp, asyncio, pandas as pd, logging, zipfile, io, tkinter as tk
from tkinter import ttk, messagebox
from datetime import datetime, timedelta
import threading, random

# === Configuration ===
SAVE_DIR = r"C:FOLDER\FILE\PATH"
os.makedirs(SAVE_DIR, exist_ok=True)
CurrentRunName = datetime.now().strftime("%Y%m%d-%H%M%S")
LOG_PATH = os.path.join(SAVE_DIR, "process_validation.log")
logging.basicConfig(filename=LOG_PATH, level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

csv_url = 'https://www.accessdata.fda.gov/MAUDE/ftparea/patientproblemcode.zip'

# === GUI Setup ===
Root = tk.Tk()
Root.title("FDA Data Processor")
Root.configure(bg='light blue')
Progress_var = tk.DoubleVar()
Status_label = ttk.Label(Root, text="Idle")
Status_label.grid(row=4, columnspan=2, pady=5)

ttk.Label(Root, text="Start Date (YYYY-MM-DD):").grid(row=0, column=0, padx=10, pady=5)
Start_date_entry = ttk.Entry(Root)
Start_date_entry.grid(row=0, column=1, padx=10, pady=5)

ttk.Label(Root, text="End Date (YYYY-MM-DD):").grid(row=1, column=0, padx=10, pady=5)
End_date_entry = ttk.Entry(Root)
End_date_entry.grid(row=1, column=1, padx=10, pady=5)

Submit_button = ttk.Button(Root, text="Submit")
Submit_button.grid(row=2, columnspan=2, pady=10)

Progress_bar = ttk.Progressbar(Root, variable=Progress_var, maximum=100)
Progress_bar.grid(row=3, columnspan=2, padx=10, pady=10)

# === Download ZIP ===
def download_csv_data(retries=3):
headers = {"User-Agent": "Mozilla/5.0"}
for attempt in range(1, retries + 1):
try:
response = requests.get(csv_url, headers=headers, timeout=20)
if response.status_code == 200:
with zipfile.ZipFile(io.BytesIO(response.content)) as z:
with z.open(z.namelist()[0]) as f:
return pd.read_csv(f, delimiter='|')
except Exception as e:
logging.warning(f"ZIP download error (attempt {attempt}): {e}")
return pd.DataFrame()

# === Async Fetch per Day with Logging ===
async def fetch_day_data(session, date_str, max_retries=3):
url = f'https://api.fda.gov/device/event.json?search=date_received:[{date_str}+TO+{date_str}]&limit=1000'
results_df = pd.DataFrame()

async def get_with_retries(url):
for attempt in range(1, max_retries + 1):
try:
async with session.get(url, headers={"Accept-Encoding": "gzip"}) as response:
logging.info(f"{date_str}: Attempt {attempt}, HTTP {response.status}")
if response.status == 404:
logging.warning(f"{date_str}: No data found (404).")
return None, 404
if response.status == 429:
raw_retry = response.headers.get("Retry-After", "5")
retry_after = min(int(raw_retry), 30)
jitter = random.uniform(1, 3)
logging.warning(f"{date_str}: 429. Capped retry {retry_after}s + jitter {jitter:.1f}s (attempt {attempt})")
await asyncio.sleep(retry_after + jitter)
continue
if response.status != 200:
logging.warning(f"{date_str}: HTTP {response.status} (attempt {attempt})")
continue
return await response.json(), 200
except Exception as e:
logging.error(f"{date_str}: Exception on attempt {attempt}: {e}")
return None, -1

data, status = await get_with_retries(url)
if status != 200 or not data:
logging.error(f"{date_str}: Final status {status}. No data returned.")
return results_df, -1, 0

expected = data.get("meta", {}).get("results", {}).get("total", 0)
results_df = pd.DataFrame(data.get("results", []))
logging.info(f"{date_str}: Expected {expected}. Page 1 got {len(results_df)}")

next_link = data.get("Link")
page_count = 1
while next_link:
logging.info(f"{date_str}: Fetching page {page_count + 1}")
next_data, next_status = await get_with_retries(next_link)
if not next_data or next_status != 200:
logging.warning(f"{date_str}: Pagination stopped at page {page_count + 1}")
break
more = pd.DataFrame(next_data.get("results", []))
if not more.empty:
results_df = pd.concat([results_df, more], ignore_index=True)
next_link = next_data.get("Link")
page_count += 1

logging.info(f"{date_str}: Retrieved {len(results_df)} total records across {page_count} page(s)")
return results_df, expected, len(results_df)

# === Batch Async Fetch ===
async def fetch_date_set(dates, label="Main Run"):
results_list, daily_summary = [], []
async with aiohttp.ClientSession() as session:
for i in range(0, len(dates), 10):
batch = dates[i:i+10]
tasks = [fetch_day_data(session, d) for d in batch]
results = await asyncio.gather(*tasks)
await asyncio.sleep(2)
for j, date_str in enumerate(batch):
try:
df, expected, retrieved = results[j]
except Exception as e:
logging.error(f"{date_str}: Batch error: {e}")
df, expected, retrieved = pd.DataFrame(), -1, 0
if not df.empty:
results_list.append(df)
status = "Success" if expected == retrieved else ("Partial" if retrieved > 0 else "Failed")
daily_summary.append({"Date": date_str, "Expected": expected, "Retrieved": retrieved, "Status": status})
Root.after(0, Progress_var.set, len(daily_summary) / len(dates) * 100)
return results_list, daily_summary

# === Main Run Submission ===
def on_submit():
try:
start_date = datetime.strptime(Start_date_entry.get(), '%Y-%m-%d')
end_date = datetime.strptime(End_date_entry.get(), '%Y-%m-%d')
Progress_var.set(0)

def run():
try:
Status_label.config(text="⏳ Fetching data...")
Root.update()
dates = [(start_date + timedelta(days=i)).strftime('%Y%m%d') for i in range((end_date - start_date).days + 1)]
all_dfs, all_summary = asyncio.run(fetch_date_set(dates))

failed_dates = [row["Date"] for row in all_summary if row["Status"] == "Failed"]
if failed_dates:
Status_label.config(text=" Retrying failed dates...")
Root.update()
retry_dfs, retry_summary = asyncio.run(fetch_date_set(failed_dates, label="Retry"))
all_dfs.extend(retry_dfs)
for retry_row in retry_summary:
for idx, row in enumerate(all_summary):
if row["Date"] == retry_row["Date"]:
all_summary[idx] = retry_row
break

pd.DataFrame(all_summary).to_csv(os.path.join(SAVE_DIR, f"{CurrentRunName}_daily_summary.csv"), index=False, encoding='utf-8-sig')

Status_label.config(text="️ Processing ZIP data...")
Root.update()
api_df = pd.concat(all_dfs, ignore_index=True) if all_dfs else pd.DataFrame()
if api_df.empty:
messagebox.showerror("Error", "No API data retrieved.")
return

api_df.columns = api_df.columns.str.lower().str.strip()
zip_df = download_csv_data()
if zip_df.empty:
messagebox.showerror("Error", "ZIP download failed.")
return
zip_df.columns = zip_df.columns.str.lower().str.strip()

if "mdr_report_key" not in api_df.columns or "mdr_report_key" not in zip_df.columns:
messagebox.showerror("Error", "mdr_report_key column missing in data.")
return

api_df["mdr_report_key"] = api_df["mdr_report_key"].astype(str)
zip_df["mdr_report_key"] = zip_df["mdr_report_key"].astype(str)
api_keys = set(api_df["mdr_report_key"])
zip_keys = set(zip_df["mdr_report_key"])
matched_keys = api_keys & zip_keys
unmatched_keys = api_keys - zip_keys

api_df.to_parquet(os.path.join(SAVE_DIR, "api_results.parquet"), index=False)
filtered_df = zip_df[zip_df["mdr_report_key"].isin(matched_keys)]
filtered_df.to_parquet(os.path.join(SAVE_DIR, "PatientProblemData.parquet"), index=False)

with open(os.path.join(SAVE_DIR, "UnmatchedKeys.txt"), "w") as f:
for key in sorted(unmatched_keys):
f.write(f"{key}\n")

Status_label.config(text="✅ Done!")
Progress_var.set(100)
messagebox.showinfo("Success", f"Process complete! Files saved to {SAVE_DIR}")

except Exception as e:
logging.error(f"Run error: {e}")
messagebox.showerror("Error", f"Unexpected error during run: {e}")
Status_label.config(text="❌ Error during processing.")
Progress_var.set(0)

threading.Thread(target=run, daemon=True).start()

except Exception as e:
logging.error(f"Submission error: {e}")
messagebox.showerror("Error", f"Invalid input: {e}")
Status_label.config(text="❌ Invalid input.")
Progress_var.set(0)

# === Final Binding ===
Submit_button.config(command=on_submit)
Root.mainloop()

Continue reading...