← Back to Libraries

Requests

Most popular HTTP library for Python with 'HTTP for Humans' concept providing simple and intuitive API. Comprehensive features including session management, SSL verification, automatic redirects, authentication, cookie handling, and file uploads.

HTTP ClientPythonSimpleAuthenticationSession Management

Library

Requests

Overview

Requests is "a simple and user-friendly HTTP library for Python" developed as the most popular HTTP client library in the Python ecosystem. With the concept of "HTTP for Humans," it provides complex HTTP processing through a simple and intuitive API. Comprehensively supporting features necessary for web API integration such as authentication, session management, SSL certificate verification, and automatic JSON/XML parsing, it has established itself as the de facto standard library for Python developers.

Details

Requests 2025 edition continues to maintain its solid position as the definitive Python HTTP communication solution. With over 15 years of development experience, it boasts mature APIs and excellent stability, being widely adopted in major web framework environments such as Django, Flask, and FastAPI. Designed with emphasis on simple and readable code writing, it enables HTTP communication implementation in a Pythonic and natural way. It provides rich features that meet enterprise-level HTTP communication requirements including session management, authentication systems, proxy support, SSL/TLS configuration, and streaming processing.

Key Features

  • Simple API: Intuitive and readable HTTP request description
  • Comprehensive Authentication Support: Support for Basic, Digest, OAuth, and custom authentication
  • Session Management: Cookie persistence and efficient connection pooling
  • Automatic Content Parsing: Automatic parsing for JSON, XML, HTML, etc.
  • Complete SSL/TLS Support: Certificate verification and client certificate support
  • Streaming Processing: Efficient processing of large files

Pros and Cons

Pros

  • Overwhelming adoption rate in Python ecosystem with abundant learning resources
  • High development efficiency and code readability through simple and intuitive API
  • Enterprise-level support through comprehensive authentication and session management
  • Rich community support and third-party extension features
  • Excellent integration with web frameworks like Django and Flask
  • Stable long-term support and backward compatibility maintenance

Cons

  • Synchronous processing-based, unsuitable for CPU-bound tasks
  • Requires dedicated libraries like aiohttp for asynchronous processing
  • Limited HTTP/2 support (requires requests-http2)
  • Performance constraints with large numbers of parallel requests
  • Relatively large bundle size due to many dependency libraries
  • Migration challenges due to end of support for older Python versions

Reference Pages

Code Examples

Installation and Basic Setup

# Install Requests
pip install requests

# Security-enhanced version (recommended)
pip install requests[security]

# Complete installation including dependencies
pip install requests[socks]

# Verification in Python environment
python -c "import requests; print(requests.__version__)"

Basic Requests (GET/POST/PUT/DELETE)

import requests

# Basic GET request
response = requests.get('https://api.example.com/users')
print(response.status_code)  # 200
print(response.headers['content-type'])  # application/json
print(response.text)  # Response text
data = response.json()  # Parse as JSON
print(data)

# GET request with query parameters
params = {'page': 1, 'limit': 10, 'sort': 'created_at'}
response = requests.get('https://api.example.com/users', params=params)
print(response.url)  # https://api.example.com/users?page=1&limit=10&sort=created_at

# POST request (sending JSON)
user_data = {
    'name': 'John Doe',
    'email': '[email protected]',
    'age': 30
}

response = requests.post(
    'https://api.example.com/users',
    json=user_data,  # Automatically sets Content-Type: application/json
    headers={'Authorization': 'Bearer your-token'}
)

if response.status_code == 201:
    created_user = response.json()
    print(f"User created: ID={created_user['id']}")
else:
    print(f"Error: {response.status_code} - {response.text}")

# POST request (sending form data)
form_data = {'username': 'testuser', 'password': 'secret123'}
response = requests.post('https://api.example.com/login', data=form_data)

# PUT request (data update)
updated_data = {'name': 'Jane Doe', 'email': '[email protected]'}
response = requests.put(
    'https://api.example.com/users/123',
    json=updated_data,
    headers={'Authorization': 'Bearer your-token'}
)

# DELETE request
response = requests.delete(
    'https://api.example.com/users/123',
    headers={'Authorization': 'Bearer your-token'}
)

if response.status_code == 204:
    print("User deleted successfully")

# Detailed response attribute inspection
print(f"Status code: {response.status_code}")
print(f"Reason: {response.reason}")
print(f"Headers: {response.headers}")
print(f"Encoding: {response.encoding}")
print(f"Request URL: {response.url}")
print(f"History: {response.history}")  # Redirect history

Advanced Configuration and Customization (Headers, Authentication, Timeout, etc.)

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# Custom header configuration
headers = {
    'User-Agent': 'MyApp/1.0 (Python Requests)',
    'Accept': 'application/json',
    'Accept-Language': 'en-US,ja-JP',
    'X-API-Version': 'v2',
    'X-Request-ID': 'req-12345'
}

response = requests.get('https://api.example.com/data', headers=headers)

# Basic authentication
response = requests.get(
    'https://api.example.com/private',
    auth=('username', 'password')
)

# Or explicitly use HTTPBasicAuth
auth = HTTPBasicAuth('username', 'password')
response = requests.get('https://api.example.com/private', auth=auth)

# Digest authentication
digest_auth = HTTPDigestAuth('username', 'password')
response = requests.get('https://api.example.com/digest', auth=digest_auth)

# Bearer Token authentication
headers = {'Authorization': 'Bearer your-jwt-token'}
response = requests.get('https://api.example.com/protected', headers=headers)

# Timeout configuration
try:
    # Connection timeout 5 seconds, read timeout 10 seconds
    response = requests.get('https://api.example.com/slow', timeout=(5, 10))
    
    # Overall timeout of 15 seconds
    response = requests.get('https://api.example.com/data', timeout=15)
    
except requests.exceptions.Timeout:
    print("Request timed out")

# SSL configuration and client certificates
response = requests.get(
    'https://secure-api.example.com/data',
    cert=('/path/to/client.cert', '/path/to/client.key'),  # Client certificate
    verify='/path/to/ca-bundle.crt'  # CA certificate bundle
)

# Disable SSL certificate verification (development only)
response = requests.get('https://self-signed.example.com/', verify=False)

# Proxy configuration
proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'http://proxy.example.com:8080'
}

response = requests.get('https://api.example.com/data', proxies=proxies)

# Authenticated proxy
proxies = {
    'http': 'http://user:[email protected]:8080',
    'https': 'http://user:[email protected]:8080'
}

# Cookie configuration
cookies = {'session_id': 'abc123', 'user_pref': 'dark_mode'}
response = requests.get('https://api.example.com/user-data', cookies=cookies)

# Redirect control
response = requests.get(
    'https://api.example.com/redirect',
    allow_redirects=False,  # Disable redirects
    max_redirects=5  # Maximum number of redirects
)

Error Handling and Retry Functionality

import requests
from requests.exceptions import (
    ConnectionError, Timeout, RequestException, 
    HTTPError, TooManyRedirects
)
import time
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

# Comprehensive error handling
def safe_request(url, **kwargs):
    try:
        response = requests.get(url, **kwargs)
        
        # HTTP status code check
        response.raise_for_status()  # Raises HTTPError for 4xx/5xx
        
        return response
        
    except ConnectionError as e:
        print(f"Connection error: {e}")
        print("Please check your network connection")
    except Timeout as e:
        print(f"Timeout error: {e}")
        print("Request timed out")
    except HTTPError as e:
        print(f"HTTP error: {e}")
        print(f"Status code: {response.status_code}")
        print(f"Response: {response.text}")
    except TooManyRedirects as e:
        print(f"Redirect error: {e}")
        print("Too many redirects")
    except RequestException as e:
        print(f"Request error: {e}")
        print("Unexpected error occurred")
    
    return None

# Usage example
response = safe_request('https://api.example.com/data', timeout=10)
if response:
    data = response.json()
    print(data)

# Manual retry implementation
def request_with_retry(url, max_retries=3, backoff_factor=1, **kwargs):
    for attempt in range(max_retries + 1):
        try:
            response = requests.get(url, **kwargs)
            response.raise_for_status()
            return response
            
        except requests.exceptions.RequestException as e:
            if attempt == max_retries:
                print(f"Maximum attempts reached: {e}")
                raise
            
            wait_time = backoff_factor * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed. Retrying in {wait_time} seconds...")
            time.sleep(wait_time)

# Usage example
try:
    response = request_with_retry(
        'https://api.example.com/unstable',
        max_retries=3,
        backoff_factor=1,
        timeout=10
    )
    print("Request successful:", response.status_code)
except requests.exceptions.RequestException as e:
    print("Finally failed:", e)

# Advanced retry configuration using urllib3's Retry class
retry_strategy = Retry(
    total=5,  # Total number of attempts
    status_forcelist=[429, 500, 502, 503, 504],  # Status codes to retry
    method_whitelist=["HEAD", "GET", "OPTIONS"],  # Methods to retry
    backoff_factor=1,  # Backoff factor
    respect_retry_after_header=True  # Respect Retry-After header
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)

# Request using session
response = session.get('https://api.example.com/data', timeout=10)

# Status code-specific handling
response = requests.get('https://api.example.com/status-check')

if response.status_code == 200:
    print("Success: ", response.json())
elif response.status_code == 401:
    print("Authentication error: Please check your token")
elif response.status_code == 403:
    print("Permission error: Access denied")
elif response.status_code == 404:
    print("Not found: Resource does not exist")
elif response.status_code == 429:
    print("Rate limit: Please wait before retrying")
elif response.status_code >= 500:
    print("Server error: Problem on server side")
else:
    print(f"Unexpected status: {response.status_code}")

Concurrent Processing and Asynchronous Requests

import requests
import concurrent.futures
from threading import Thread
import queue
import time

# Parallel fetching of multiple URLs
def fetch_url(url):
    try:
        response = requests.get(url, timeout=10)
        return {
            'url': url,
            'status_code': response.status_code,
            'content_length': len(response.content),
            'success': True
        }
    except requests.exceptions.RequestException as e:
        return {
            'url': url,
            'error': str(e),
            'success': False
        }

# Parallel processing using ThreadPoolExecutor
urls = [
    'https://api.example.com/users',
    'https://api.example.com/posts',
    'https://api.example.com/comments',
    'https://api.example.com/categories'
]

def parallel_fetch_threadpool(urls, max_workers=5):
    results = []
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Submit all URLs for parallel execution
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        
        for future in concurrent.futures.as_completed(future_to_url):
            result = future.result()
            results.append(result)
            
            if result['success']:
                print(f"Success: {result['url']} - {result['status_code']}")
            else:
                print(f"Failed: {result['url']} - {result['error']}")
    
    return results

# Usage example
results = parallel_fetch_threadpool(urls)
successful_results = [r for r in results if r['success']]
print(f"Success: {len(successful_results)}/{len(urls)}")

# Efficient parallel processing with sessions
def fetch_with_session(session, url):
    try:
        response = session.get(url, timeout=10)
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error {url}: {e}")
        return None

def parallel_fetch_with_session(urls, max_workers=5):
    results = []
    
    # Sessions share connection pools for efficiency
    session = requests.Session()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(fetch_with_session, session, url) for url in urls]
        
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result:
                results.append(result)
    
    session.close()
    return results

# Pagination-aware progressive data fetching
def fetch_all_pages(base_url, headers=None, max_pages=None):
    all_data = []
    page = 1
    session = requests.Session()
    
    if headers:
        session.headers.update(headers)
    
    while True:
        try:
            params = {'page': page, 'per_page': 100}
            response = session.get(base_url, params=params, timeout=10)
            response.raise_for_status()
            
            data = response.json()
            
            if not data or (isinstance(data, list) and len(data) == 0):
                break  # End if no data
            
            if isinstance(data, dict) and 'items' in data:
                items = data['items']
                all_data.extend(items)
                
                # End if no next page
                if not data.get('has_more', True) or len(items) == 0:
                    break
            else:
                all_data.extend(data)
            
            print(f"Page {page} completed: {len(data if isinstance(data, list) else data.get('items', []))} items")
            page += 1
            
            # Check maximum page limit
            if max_pages and page > max_pages:
                break
            
            # Wait to reduce API load
            time.sleep(0.1)
            
        except requests.exceptions.RequestException as e:
            print(f"Error on page {page}: {e}")
            break
    
    session.close()
    print(f"Total items fetched: {len(all_data)}")
    return all_data

# Usage example
headers = {'Authorization': 'Bearer your-token'}
all_posts = fetch_all_pages('https://api.example.com/posts', headers=headers)

# Duplicate request prevention and caching
class RequestCache:
    def __init__(self):
        self.cache = {}
        self.pending = {}
    
    def get_or_fetch(self, url, **kwargs):
        if url in self.cache:
            print(f"Cache hit: {url}")
            return self.cache[url]
        
        if url in self.pending:
            print(f"Duplicate request in progress: {url}")
            return None
        
        try:
            self.pending[url] = True
            response = requests.get(url, **kwargs)
            response.raise_for_status()
            
            result = response.json()
            self.cache[url] = result
            return result
            
        finally:
            self.pending.pop(url, None)

# Cache usage example
cache = RequestCache()
data1 = cache.get_or_fetch('https://api.example.com/users/123')
data2 = cache.get_or_fetch('https://api.example.com/users/123')  # Retrieved from cache

Framework Integration and Practical Examples

import requests
from requests.auth import AuthBase
import json
import os
from datetime import datetime, timedelta

# API client for Django/Flask integration
class APIClient:
    def __init__(self, base_url, token=None):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'User-Agent': 'MyApp/1.0'
        })
        
        if token:
            self.session.headers['Authorization'] = f'Bearer {token}'
    
    def get(self, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return self._make_request('GET', url, **kwargs)
    
    def post(self, endpoint, data=None, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return self._make_request('POST', url, json=data, **kwargs)
    
    def put(self, endpoint, data=None, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return self._make_request('PUT', url, json=data, **kwargs)
    
    def delete(self, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return self._make_request('DELETE', url, **kwargs)
    
    def _make_request(self, method, url, **kwargs):
        try:
            response = self.session.request(method, url, timeout=30, **kwargs)
            response.raise_for_status()
            
            if response.content:
                return response.json()
            return None
            
        except requests.exceptions.RequestException as e:
            print(f"API Error: {method} {url} - {e}")
            raise
    
    def close(self):
        self.session.close()

# Usage example
client = APIClient('https://api.example.com/v1', token='your-jwt-token')

try:
    # Get user list
    users = client.get('users', params={'page': 1, 'limit': 50})
    
    # Create new user
    new_user = client.post('users', data={
        'name': 'John Doe',
        'email': '[email protected]'
    })
    
    # Update user
    updated_user = client.put(f'users/{new_user["id"]}', data={
        'name': 'Jane Doe'
    })
    
finally:
    client.close()

# Auto-refresh authentication client
class AuthRefreshClient:
    def __init__(self, base_url, client_id, client_secret):
        self.base_url = base_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
        self.token_expires_at = None
        self.session = requests.Session()
    
    def _ensure_valid_token(self):
        if (not self.access_token or 
            not self.token_expires_at or 
            datetime.now() >= self.token_expires_at):
            self._refresh_token()
    
    def _refresh_token(self):
        auth_data = {
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'grant_type': 'client_credentials'
        }
        
        response = requests.post(
            f"{self.base_url}/oauth/token",
            data=auth_data,
            timeout=10
        )
        response.raise_for_status()
        
        token_data = response.json()
        self.access_token = token_data['access_token']
        expires_in = token_data.get('expires_in', 3600)
        self.token_expires_at = datetime.now() + timedelta(seconds=expires_in)
        
        self.session.headers['Authorization'] = f'Bearer {self.access_token}'
    
    def request(self, method, endpoint, **kwargs):
        self._ensure_valid_token()
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return self.session.request(method, url, **kwargs)

# Custom authentication class
class APIKeyAuth(AuthBase):
    def __init__(self, api_key, header_name='X-API-Key'):
        self.api_key = api_key
        self.header_name = header_name
    
    def __call__(self, r):
        r.headers[self.header_name] = self.api_key
        return r

# Usage example
api_auth = APIKeyAuth('your-api-key-here')
response = requests.get('https://api.example.com/data', auth=api_auth)

# File upload
def upload_file(file_path, upload_url, additional_fields=None):
    """Upload file in multipart format"""
    
    files = {'file': open(file_path, 'rb')}
    data = additional_fields or {}
    
    try:
        response = requests.post(
            upload_url,
            files=files,
            data=data,
            headers={'Authorization': 'Bearer your-token'},
            timeout=300  # 5-minute timeout
        )
        response.raise_for_status()
        return response.json()
        
    finally:
        files['file'].close()

# Usage example
upload_result = upload_file(
    '/path/to/document.pdf',
    'https://api.example.com/upload',
    additional_fields={'category': 'documents', 'public': 'false'}
)

# Streaming download
def download_large_file(url, local_filename, chunk_size=8192):
    """Download large file using streaming"""
    
    with requests.get(url, stream=True, timeout=30) as response:
        response.raise_for_status()
        
        # Get file size (if available)
        total_size = int(response.headers.get('content-length', 0))
        downloaded = 0
        
        with open(local_filename, 'wb') as f:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if chunk:  # Filter
                    f.write(chunk)
                    downloaded += len(chunk)
                    
                    # Progress display
                    if total_size > 0:
                        progress = (downloaded / total_size) * 100
                        print(f"Download progress: {progress:.1f}%", end='\r')
        
        print(f"\nDownload completed: {local_filename}")

# Usage example
download_large_file(
    'https://api.example.com/files/large-dataset.zip',
    '/tmp/dataset.zip'
)

# Webhook receiving server (Flask integration example)
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    """Handle webhook reception from external APIs"""
    
    # Sender verification
    expected_token = os.environ.get('WEBHOOK_SECRET')
    received_token = request.headers.get('X-Webhook-Token')
    
    if not received_token or received_token != expected_token:
        return jsonify({'error': 'Unauthorized'}), 401
    
    # Process webhook data
    webhook_data = request.json
    
    # Send notification to external API
    try:
        notification_response = requests.post(
            'https://internal-api.example.com/notifications',
            json={
                'event': webhook_data.get('event'),
                'timestamp': datetime.now().isoformat(),
                'data': webhook_data
            },
            headers={'Authorization': 'Bearer internal-token'},
            timeout=10
        )
        notification_response.raise_for_status()
        
    except requests.exceptions.RequestException as e:
        print(f"Notification send error: {e}")
        return jsonify({'error': 'Notification failed'}), 500
    
    return jsonify({'status': 'processed'}), 200