Concurrent Request Python

concurrent.py

# Ref: https://stackoverflow.com/a/68583332/5994461
import logging
import random
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests
from requests.structures import CaseInsensitiveDict

THREAD_POOL = 16

# This is how to create a reusable connection pool with python requests.
session = requests.Session()
session.mount(
    "https://",
    requests.adapters.HTTPAdapter(
        pool_maxsize=THREAD_POOL, max_retries=3, pool_block=True
    ),
)


def post(url):
    headers = CaseInsensitiveDict()
    data = {
        "isPublic": True,
        "name": f"concurrent-{random.randint(0, 100)}",
        "storageLimit": 2147483648,  # Change to the maximum size of allowed quota (currently 2GB)
        "unitSuffix": "GB",
    }
    # user-token cookie
    headers[
        "Cookie"
    ] = "user-token=[YOUR_TOKEN_HERE]"
    response = session.post(
        url,
        json=data,
        headers=headers,
    )
    logging.info(
        "POST request was completed in %s seconds [%s]",
        response.elapsed.total_seconds(),
        response.url,
    )
    if response.status_code != 200:
        logging.error(
            "POST request failed, error code %s [%s]",
            response.status_code,
            response.url,
        )
    if 500 <= response.status_code < 600:
        # server is overloaded? give it a break
        time.sleep(3)
    return response


def download(urls):
    with ThreadPoolExecutor(max_workers=THREAD_POOL) as executor:
        # wrap in a list() to wait for all requests to complete
        for response in list(executor.map(post, urls)):
            if response.status_code == 201:
                print(response.content)


def main():
    logging.basicConfig(
        format="%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s",
        level=logging.INFO,
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    # List of request URLs to perform concurrently
    urls = [
        "API_URL_HERE",
        "API_URL_HERE",
        "API_URL_HERE",
        "API_URL_HERE",
        "API_URL_HERE",
    ]

    download(urls)


if __name__ == "__main__":
    main()