# Dive into requests

By [Aki](https://paragraph.com/@aki) · 2022-04-04

---

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 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](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)

---

*Originally published on [Aki](https://paragraph.com/@aki/dive-into-requests)*