The first thing to point out is that you should understand that fetching all posts with a tag is going to result in huge quota usage for any relatively popular tag , as well as take a significant amount of time.
For example, ATTOW, the swift tag has 314130 questions, meaning you will have to make 3142 requests of 100 items to get all the posts once, a whopping ~1/3 of the daily extended quota (see #2.6 in this regards below).
I assume you are already aware of this, but it is still extremely important to note as, IIRC, there isn't currently a way to fetch more than 100 posts at once.
As for the implementation, there is a couple of things that do not seem right about it:
The page_size parameter naming is misleading. The pagesize is correctly hardcoded by you to always be 100, whereas the query parameter you actually set is page. While it works correctly, reading your code has a mental overhead of concluding that fetch_question(page_count) is actually correct.
Same goes for the page_count name — it implies the argument controls the number of pages requested, but in reality it just sets the page query parameter.
Your filter includes way too many fields for what the get_all_questions function seems to be doing. If this is a simplified example, and you actually use the items list to accumulate the posts in the real code, here are a few things to address:
total field is a convenience field for requests that just need a total number of items in a collection and nothing else. Getting it is expensive, but you are requesting it on every paginated request.
Once you fetch all the items you need, just len() the accumulated list to get the total_fetched stat (which will also save you the need to track items_count [you also seem to have a typo — count_ should be items_count instead]), and then (or before the paginated requests) fetch the API once with only the total field in the filter (and obviously with the error handling-related ones) to get the total according to the API.
The implementation could be simplified with recursive calls instead of a while loop (no need to overwrite has_more and quota_remain or even have those vars in the first place).
You do not really need to throttle yourself to 1 page per second. While the API does not guarantee the stated 30 requests per 1 second (and in fact often starts to throw tantrums on about 10-15 requests a second), fetching relatively popular tags would be atrociously slow for your end users (might not matter much if it's just a scheduled job as time.sleep(3600) implies, but still something to keep in mind).
Sometimes you can get rate-limited without even violating or hitting the backoff parameter. You might want to have a try...except statement in fetch_question just in case (and generally for unrelated issues).
The backoff field, when present, is guaranteed by the API to be an integer — while nothing stops you from proofing the program against the API going mad, at the point of it violating explicit contracts you'd want to know that as soon as possible rather than trying to auto-correct.
Unless this is, again, just a minimal example, you should register an API key for the app to get the 10K requests quota as it will quickly use up the 300 unauthenticated requests quota.
Other than the above, your implementation seems to be doing what it is supposed to do, including handing of the quota_remaining, backoff, and general throttling, so happy fetching!
Based on the notes above, though, I'd still consider tweaking it somewhat like this (implementing some of the suggestions above is left as an exercise for the reader):
class Fetch_Result(TypedDict):
api_total: int
items: list[dict]
total_fetched: int
class API_Res(TypedDict):
backoff: Optional[int]
has_more: Optional[bool]
items: list[dict]
quota_remaining: Optional[int]
def fetch_questions(tag: str, page: int = 1) -> API_Res:
filter = "!*L2iagzJ_cnEq*JT"
api_res = f"https://api.stackexchange.com/2.3/questions?page={int(page)}&pagesize=100&order=desc&sort=activity&tagged={tag}&site=stackoverflow&filter={filter}"
response = requests.get(api_res)
return response.json()
def fetch_totals(tag: str) -> int:
api_res = f"https://api.stackexchange.com/2.3/questions?tagged={tag}&site=stackoverflow&filter=total"
response = requests.get(api_res)
res = response.json()
if 'backoff' in res:
time.sleep(res['backoff']+1)
return fetch_totals(tag)
return res["total"]
def get_all_questions(tag: str, page: int = 1) -> Fetch_Result:
res = fetch_questions(tag, page)
items = res["items"]
if not res['has_more']:
return {
"total_fetched": len(items),
"items": items,
"api_total": fetch_totals(tag),
}
if res['quota_remaining'] == 1:
"""If exhausted quota then sleep for one day
to reset the quota limit"""
time.sleep(3600)
if 'backoff' in res:
"""Respect API backoff time"""
time.sleep(res['backoff']+1)
time.sleep(1)
next_res = get_all_questions(tag, page + 1)
all_items = items + next_res["items"]
return {
"total_fetched": len(all_items),
"items": all_items,
"api_total": next_res["api_total"],
}
