What the interviewer looks for: always a timeout (requests waits forever by default), a Session for connection pooling, retry only on idempotent methods and on 5xx, respecting 429 Retry-After, our own exceptions so the rest of the code does not depend on
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 | # ====================================================================== # Q: Write a production-ready client for an external API. # # What the interviewer looks for: always a timeout (requests waits forever by # default), a Session for connection pooling, retry only on idempotent methods # and on 5xx, respecting 429 Retry-After, our own exceptions so the rest of the # code does not depend on requests, and never calling an external API inside a # DB transaction. # ====================================================================== import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class ExternalAPIError(Exception): ... class RateLimited(ExternalAPIError): def __init__(self, retry_after): self.retry_after = retry_after class WeatherClient: BASE_URL = 'https://api.example.com/v2' def __init__(self, api_key, timeout=(3.05, 10)): # (connect, read) self.session = requests.Session() # connection pooling retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[502, 503, 504], allowed_methods=['GET']) # idempotent only self.session.mount('https://', HTTPAdapter(max_retries=retry)) self.session.headers['Authorization'] = f'Bearer {api_key}' self.timeout = timeout def forecast(self, city): r = self.session.get(f'{self.BASE_URL}/forecast', params={'q': city}, timeout=self.timeout) if r.status_code == 429: raise RateLimited(int(r.headers.get('Retry-After', 60))) if not r.ok: raise ExternalAPIError(f'{r.status_code}: {r.text[:200]}') return r.json() |