“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 example

A normal usage example of requests may look like below snippets.

There’re several problems in the snippets:

1. Use while loop to retry requests. You may have to write those codes for hundreds of times in different functions.

2. 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.

3. 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.

4. 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 calling

Actually, 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-Key

self defined requests

import 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 check

def 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 retry

def 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)

Final

Full 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.org

import 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)