Friday, November 6, 2026

Making a request to a RESTful API using Python

Python requests: Examples of Every Type

The Python requests library is the standard tool used to interact with web APIs by sending HTTP requests. Because it is a third-party library, it must be installed prior to use via pip install requests.

Core HTTP Methods
HTTP MethodPython SyntaxPrimary Use Case
GETrequests.get(url)   Fetching read-only data from an API.
POSTrequests.post(url, json=data)   Sending data to create a new resource.
PUTrequests.put(url, json=data)   Updating or replacing an existing resource.
PATCHrequests.patch(url, json=data)   Modifying specific parts of an existing resource.
DELETErequests.delete(url)   Deleting a specified resource.
Making a request to a RESTful API using Python



1. HTTP methods

import requests
base = "https://httpbin.org"

requests.get(f"{base}/get")
requests.post(f"{base}/post", json={"a": 1})
requests.put(f"{base}/put", json={"a": 2})
requests.patch(f"{base}/patch", json={"a": 3})
requests.delete(f"{base}/delete")
requests.head(f"{base}/get")        # headers only, no body
requests.options(f"{base}/get")     # allowed methods in r.headers["Allow"]

# Generic form
requests.request("GET", f"{base}/get")

2. Query parameters

# Dict
requests.get(url, params={"q": "python", "page": 2})

# List values -> ?tag=a&tag=b
requests.get(url, params={"tag": ["a", "b"]})

# List of tuples (allows repeated keys, preserves order)
requests.get(url, params=[("sort", "asc"), ("sort", "date")])

3. Request bodies

# JSON
requests.post(url, json={"name": "Asha", "tags": ["x", "y"]})

# Form-encoded (application/x-www-form-urlencoded)
requests.post(url, data={"user": "asha", "pw": "secret"})

# Raw string / bytes
requests.post(url, data="plain text body", headers={"Content-Type": "text/plain"})
requests.post(url, data=b"\x00\x01\x02", headers={"Content-Type": "application/octet-stream"})

# XML
xml = "<user><name>Asha</name></user>"
requests.post(url, data=xml, headers={"Content-Type": "application/xml"})

# Generator (chunked upload)
def gen():
    yield b"part1"
    yield b"part2"
requests.post(url, data=gen())

4. File uploads (multipart)

# Single file
with open("report.pdf", "rb") as f:
    requests.post(url, files={"file": f})

# Custom filename and content type
with open("data.csv", "rb") as f:
    requests.post(url, files={"file": ("my_data.csv", f, "text/csv")})

# Multiple files
files = [
    ("images", ("a.png", open("a.png", "rb"), "image/png")),
    ("images", ("b.png", open("b.png", "rb"), "image/png")),
]
requests.post(url, files=files)

# Files + extra form fields
with open("doc.txt", "rb") as f:
    requests.post(url, files={"file": f}, data={"description": "My doc"})

5. Headers

headers = {
    "User-Agent": "my-app/1.0",
    "Accept": "application/json",
    "Accept-Language": "en-IN",
    "X-Custom-Header": "value",
}
r = requests.get(url, headers=headers)
print(r.request.headers)   # what was actually sent

6. Authentication

from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# Basic
requests.get(url, auth=("user", "pass"))
requests.get(url, auth=HTTPBasicAuth("user", "pass"))

# Digest
requests.get(url, auth=HTTPDigestAuth("user", "pass"))

# Bearer token
requests.get(url, headers={"Authorization": "Bearer YOUR_TOKEN"})

# API key in header
requests.get(url, headers={"X-API-Key": "abc123"})

# API key in query
requests.get(url, params={"api_key": "abc123"})

# Custom auth class
class TokenAuth(requests.auth.AuthBase):
    def __init__(self, token):
        self.token = token
    def __call__(self, r):
        r.headers["Authorization"] = f"Token {self.token}"
        return r

requests.get(url, auth=TokenAuth("abc"))

7. Reading responses

r = requests.get(url)

r.status_code          # 200
r.ok                   # True if < 400
r.reason               # "OK"
r.url                  # final URL
r.text                 # decoded string
r.content              # raw bytes
r.json()               # parsed JSON
r.headers              # case-insensitive dict
r.cookies              # response cookies
r.encoding             # e.g. "utf-8" (settable: r.encoding = "latin-1")
r.elapsed              # timedelta
r.history              # list of redirect responses
r.request              # the PreparedRequest
r.raw                  # raw urllib3 response (needs stream=True)

8. Status code checks

r = requests.get(url)

if r.status_code == requests.codes.ok:
    print("OK")
elif r.status_code == requests.codes.not_found:
    print("404")

r.raise_for_status()   # raises HTTPError for 4xx/5xx

9. Cookies

# Send cookies
requests.get(url, cookies={"session_id": "abc123"})

# Read cookies
r = requests.get(url)
print(r.cookies.get("session_id"))

# Cookie jar
jar = requests.cookies.RequestsCookieJar()
jar.set("token", "xyz", domain="example.com", path="/")
requests.get(url, cookies=jar)

10. Timeouts

requests.get(url, timeout=5)          # applies to connect and read each
requests.get(url, timeout=(3, 10))    # (connect, read)
requests.get(url, timeout=None)       # wait forever (avoid)

11. Redirects

r = requests.get(url, allow_redirects=False)   # don't follow
print(r.status_code, r.headers["Location"])

r = requests.get(url)                          # follows by default
for h in r.history:
    print(h.status_code, h.url)

12. Sessions

with requests.Session() as s:
    s.headers.update({"User-Agent": "my-app"})
    s.auth = ("user", "pass")
    s.params = {"api_key": "abc"}      # sent with every request

    s.get("https://httpbin.org/cookies/set/token/123")
    r = s.get("https://httpbin.org/cookies")
    print(r.json())                    # cookie persisted

13. Proxies

proxies = {
    "http": "http://10.10.1.10:3128",
    "https": "http://10.10.1.10:1080",
}
requests.get(url, proxies=proxies)

# With auth
proxies = {"https": "http://user:pass@10.10.1.10:3128"}

# SOCKS (pip install requests[socks])
proxies = {"https": "socks5://127.0.0.1:9050"}

14. SSL / TLS

requests.get(url)                              # verify=True by default
requests.get(url, verify="/path/to/ca.pem")    # custom CA bundle
requests.get(url, verify=False)                # disable (testing only)
requests.get(url, cert=("client.crt", "client.key"))   # client certificate

15. Streaming

# Download large file
with requests.get(url, stream=True, timeout=10) as r:
    r.raise_for_status()
    with open("big.zip", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

# Line-by-line (e.g. NDJSON or SSE-style streams)
with requests.get(url, stream=True) as r:
    for line in r.iter_lines(decode_unicode=True):
        if line:
            print(line)

16. Download with progress

r = requests.get(url, stream=True)
total = int(r.headers.get("Content-Length", 0))
done = 0
with open("file.bin", "wb") as f:
    for chunk in r.iter_content(8192):
        f.write(chunk)
        done += len(chunk)
        if total:
            print(f"\r{done/total:.0%}", end="")

17. Retries and backoff

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(
    total=5,
    backoff_factor=1,                       # 1s, 2s, 4s...
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
)
s = requests.Session()
s.mount("https://", HTTPAdapter(max_retries=retry))
s.mount("http://", HTTPAdapter(max_retries=retry))
r = s.get(url, timeout=10)

18. Exception handling

from requests.exceptions import (
    Timeout, ConnectionError, HTTPError,
    TooManyRedirects, SSLError, RequestException,
)

try:
    r = requests.get(url, timeout=5)
    r.raise_for_status()
except Timeout:
    print("Timed out")
except SSLError:
    print("SSL problem")
except ConnectionError:
    print("Network problem")
except TooManyRedirects:
    print("Redirect loop")
except HTTPError as e:
    print("HTTP error:", e.response.status_code)
except RequestException as e:
    print("Other error:", e)

19. Hooks

def log_response(r, *args, **kwargs):
    print(r.status_code, r.url)

requests.get(url, hooks={"response": log_response})

20. Prepared requests (inspect before sending)

req = requests.Request("POST", url, json={"a": 1}, headers={"X-Test": "1"})
prepped = req.prepare()
print(prepped.method, prepped.url, prepped.headers, prepped.body)

with requests.Session() as s:
    r = s.send(prepped, timeout=5)

21. Concurrency with threads

from concurrent.futures import ThreadPoolExecutor

urls = [f"https://httpbin.org/delay/1?i={i}" for i in range(10)]

def fetch(u):
    return requests.get(u, timeout=10).status_code

with ThreadPoolExecutor(max_workers=5) as ex:
    print(list(ex.map(fetch, urls)))

22. Pagination pattern

def fetch_all(url, params=None):
    results, page = [], 1
    while True:
        r = requests.get(url, params={**(params or {}), "page": page}, timeout=10)
        r.raise_for_status()
        data = r.json()
        if not data:
            break
        results.extend(data)
        page += 1
    return results

# Link-header pagination (e.g. GitHub)
r = requests.get("https://api.github.com/repos/psf/requests/issues", params={"per_page": 100})
while True:
    process(r.json())
    next_url = r.links.get("next", {}).get("url")
    if not next_url:
        break
    r = requests.get(next_url)

23. Rate-limit handling

import time

r = requests.get(url)
if r.status_code == 429:
    wait = int(r.headers.get("Retry-After", 5))
    time.sleep(wait)
    r = requests.get(url)

24. Real-world mini examples

# GitHub API
r = requests.get("https://api.github.com/users/octocat", timeout=10)
print(r.json()["public_repos"])

# POST JSON and read the echo
r = requests.post("https://httpbin.org/post", json={"hello": "world"})
print(r.json()["json"])

# Download an image
r = requests.get("https://httpbin.org/image/png")
with open("pic.png", "wb") as f:
    f.write(r.content)

# Check a link without downloading the body
print(requests.head(url, allow_redirects=True).status_code)

Quick reference: key parameters

ParameterPurpose
paramsURL query string
dataForm or raw body
jsonJSON body
filesMultipart upload
headersRequest headers
cookiesCookies to send
authAuthentication
timeoutConnect/read timeout
allow_redirectsFollow redirects
proxiesProxy servers
verifySSL verification or CA path
certClient certificate
streamDefer body download
hooksEvent callbacks


Complete REST API testing example

This example combines a GET request, timeout, error handling, and assertions.

import requests

def test_get_user():
    url = "https://jsonplaceholder.typicode.com/users/1"

    response = requests.get(url, timeout=10)
    response.raise_for_status()

    assert response.status_code == 200

    user = response.json()

    assert user["id"] == 1
    assert user["name"] == "Leanne Graham"
    assert "@" in user["email"]

    print("All assertions passed")

if __name__ == "__main__":
    test_get_user()





No comments:

Post a Comment