# Dive into requests

**Published by:** [Aki](https://paragraph.com/@aki/)
**Published on:** 2022-04-04
**URL:** https://paragraph.com/@aki/dive-into-requests

## Content

Intro“The requests library is more powerful and easier to use than the urllib.request module from the Python 3 standard library. In fact, requests is considered a model Pythonic API.”As the most well-known Python HTTP library, requests actually emphasizes the major advantage of Python, i.e., Various implementations lead to the same result. However, there’re some tricks, or so called best practices in requests which is hidden under the source codes and manuals.A normal exampleA normal usage example of requests may look like below snippets. There’re several problems in the snippets:Use while loop to retry requests. You may have to write those codes for hundreds of times in different functions.There’s no exception catching and throwing. For instance, json.loads may throw an exception if res.text is not an instance of bytes or bytearray.Only use status_code to judge whether the requests is successful, which is hard to handle tricky situations, status_code = 200, code = 400 for instance.etc.import requests import json def my_requests(data): headers = {'Auth-Pass-Key': 'abc', 'Content-Type': 'application/json'} retry_index = 0 while retry_index <= 3: res = requests.post("url", json=data, headers=headers) if res.status_code != 200: retry_index += 1 else: return json.loads(res.text) return None Unified callingActually, some codes in above snippet can be replaced by built-in function/property in requests library.requests# BEFORE res = requests.post("url", json=data, headers=headers) # AFTER res = requests.request("POST", url="url", json=data, headers=headers) In this way, you can customize your own common requests, to add logs, add wrappers, etc.JSON# BEFORE return json.loads(res.text) # AFTER return res.json() You don’t have to call json.loads yourself, instead, just res.json() is enough, which will also automatically raise JSONDecodeError under scene.# BEFORE if res.status_code != 200: # AFTER if res: # OR: res.raise_for_status() based on the __bool__ overloaded dunder method in Response object. It will return True if status_code is less than 400. The second way will raise a HTTPError with context when status_code is between 400 and 600.headers# BEFORE headers = {'Auth-Pass-Key': 'abc', 'Content-Type': 'application/json'} res = requests.post("url", json=data, headers=headers) # AFTER from requests.auth import AuthBase class AuthPass(AuthBase): def __init__(self, pass): self.pass = pass def __call__(self, r): """Attach an API token to a custom auth header.""" r.headers['Auth-Pass-Key'] = f'{self.pass}' return r res = requests.post("url", json=data, auth=AuthPass("abc")) use json=data will automatically add 'Content-Type': 'application/json' to headers. use AuthBase object is better than use headers variable which contains hard coded Auth-Pass-Keyself defined requestsimport requests @auto_retry() def common_requests(method: str, **kwargs) -> Optional[Dict]: resp: Response = requests.request(method, **kwargs) logger.info(f"{resp.status_code}, {resp.text}") if is_request_success(resp): return resp.json() # may raise JSONDecodeError return None def my_post_requests(data: Dict, auth: AuthBase) -> Optional[Dict]: return common_requests("post", json=data, auth=auth) def my_get_requests(params: Dict) -> Optional[Dict]: return common_request("GET", params=params) In above snippets, a self-encapsulated requests functions is built. You can use it for all the normal requests before. In this way, you can write logs, verify the success of requests in just one place. You can even add more features in common_requests. common_requests takes the advantages of **kwargs and @auto_retry() wrapper (will talk in next section). You can also use partialmethod to create common_post_requests and etc.self defined checkdef is_request_success(resp: Response) -> bool: resp.raise_for_status() # may raise HTTPError if 400 <= resp.status_code <= 600 if is_customize_success(resp.json()): # may raise JSONDecodeError return True else: raise RequestsFailedException(resp.text) def is_customize_success(json_resp: Dict) -> bool: if json_resp.get("code", 999) in [0, "0"]: return True if json_resp.get("status_code", 999) in [0, "0"]: return True return False A self defined check function is necessary to handle “BAD” API. You can use it all over the project rather than write them in individual requests function.auto retrydef auto_retry(times=3, low=1, high=2): def outer_wrapper(f): @functools.wraps(f) def wrapper(*args, **kwargs): retry_index: int = 0 while retry_index < times: try: return f(*args, **kwargs) except (HTTPError, RequestsFailedException, JSONDecodeError) as e: logger.error(e) time.sleep(random.uniform(low, high)) retry_index += 1 return None return wrapper return outer_wrapper This wrapper will catch errors including HTTPError, throw by res.raise_for_status(), JSONDecodeError, throw by res.json(),RequestsFailedException, self defined error, throw by is_request_success(resp). It will automatically retry requests for times, between each requests, the wrapper will wait for low-high seconds. Or, you can use Session with adapters to retry automatically.import requests from requests.adapters import HTTPAdapter from requests.exceptions import ConnectionError my_adapter = HTTPAdapter(max_retries=3) session = requests.Session() session.mount('url', my_adapter) try: session.get('url') except ConnectionError as e: logger.error(e) FinalFull set of snippet is as below. For co-current scenario, you may refer to aiohttp. See requests manual for more information: https://docs.python-requests.orgimport functools import random import time from json import JSONDecodeError from typing import Dict, List, Optional, Callable import requests from requests import Response, HTTPError from requests.auth import AuthBase from exceptions import RequestsFailedException def auto_retry(times=3, low=1, high=2) -> Callable: def outer_wrapper(f) -> Callable: @functools.wraps(f) def wrapper(*args, **kwargs) -> Optional[Dict]: retry_index: int = 0 while retry_index < times: try: return f(*args, **kwargs) except (HTTPError, RequestsFailedException, JSONDecodeError) as e: logger.error(e) time.sleep(random.uniform(low, high)) retry_index += 1 return None return wrapper return outer_wrapper @auto_retry() def common_requests(method: str, **kwargs) -> Optional[Dict]: resp: Response = requests.request(method, **kwargs) if is_request_success(resp): return resp.json() # raise JSONDecodeError return None def is_request_success(resp: Response) -> bool: resp.raise_for_status() # will raise HTTPError if 400 <= resp.status_code <= 600 if is_customize_success(resp.json()): return True else: raise RequestsFailedException(resp.text) def is_customize_success(json_resp: Dict) -> bool: if json_resp.get("code", 999) in [0, "0"]: return True if json_resp.get("status_code", 999) in [0, "0"]: return True return False class AuthPass(AuthBase): def __init__(self, pass): self.pass = pass def __call__(self, r): """Attach an API token to a custom auth header.""" r.headers['Auth-Pass-Key'] = f'{self.pass}' return r def my_post_requests(data: Dict, auth: AuthBase) -> Optional[Dict]: return common_requests("post", json=data, auth=auth) def my_get_requests(params: Dict) -> Optional[Dict]: return common_request("GET", params=params)

## Publication Information

- [Aki](https://paragraph.com/@aki/): Publication homepage
- [All Posts](https://paragraph.com/@aki/): More posts from this publication
- [RSS Feed](https://api.paragraph.com/blogs/rss/@aki): Subscribe to updates