or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

tessl/pypi-requests-unixsocket

Use requests to talk HTTP via a UNIX domain socket

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/requests-unixsocket@0.4.x

To install, run

npx @tessl/cli install tessl/pypi-requests-unixsocket@0.4.0

index.mddocs/

requests-unixsocket

A Python library that enables HTTP communication over UNIX domain sockets using the familiar requests API. It extends the popular requests library to support UNIX socket connections, making it ideal for communicating with local services like Docker daemons, uWSGI servers, and other socket-based services.

Package Information

  • Package Name: requests-unixsocket
  • Language: Python
  • Installation: pip install requests-unixsocket
  • Dependencies: requests >= 1.1

Core Imports

import requests_unixsocket

For specific components:

from requests_unixsocket import Session, monkeypatch, UnixAdapter

For testing utilities:

from requests_unixsocket.testutils import UnixSocketServerThread

Basic Usage

Using Session Class (Explicit)

import requests_unixsocket

# Create a session configured for UNIX socket communication
session = requests_unixsocket.Session()

# Make HTTP requests to a UNIX socket (Docker daemon example)
response = session.get('http+unix://%2Fvar%2Frun%2Fdocker.sock/info')
print(response.json())

# Support for all HTTP methods
response = session.post('http+unix://%2Fvar%2Frun%2Fdocker.sock/containers/create', 
                       json={'Image': 'nginx'})

Using Monkeypatch (Implicit)

import requests
import requests_unixsocket

# Temporarily patch the global requests module
with requests_unixsocket.monkeypatch():
    response = requests.get('http+unix://%2Fvar%2Frun%2Fdocker.sock/info')
    print(response.status_code)

Abstract Namespace Sockets (Linux)

import requests_unixsocket

session = requests_unixsocket.Session()

# Use \0 prefix for abstract namespace sockets
response = session.get('http+unix://\0test_socket/api/status')
print(response.text)

URL Format

UNIX socket URLs use the http+unix:// scheme with percent-encoded socket paths:

  • Regular socket: http+unix://%2Ftmp%2Fsocket.sock/path/to/endpoint
  • Abstract namespace (Linux): http+unix://\0socket_name/path/to/endpoint

The socket path is percent-encoded, and the HTTP path follows after the socket specification.

Architecture

The library provides multiple interface layers for different use cases:

  • Session Layer: High-level Session class that extends requests.Session with UNIX socket support. Provides the most familiar API for users already using requests.
  • Function Layer: Module-level HTTP functions (get, post, etc.) that work like their requests counterparts but support UNIX sockets.
  • Monkeypatch Layer: monkeypatch class that temporarily or permanently modifies the global requests module to support UNIX sockets, allowing existing code to work unchanged.
  • Adapter Layer: Low-level UnixAdapter class that handles the actual UNIX socket connections. Can be used directly for custom session configurations.
  • Testing Layer: UnixSocketServerThread utility for creating test servers on UNIX sockets.

This layered design allows the library to support everything from drop-in compatibility (via monkeypatching) to fine-grained control (via direct adapter usage).

Capabilities

Session Class

Provides a requests-compatible session for UNIX socket communication.

class Session(requests.Session):
    def __init__(self, url_scheme=DEFAULT_SCHEME, *args, **kwargs):
        """
        Create a session configured for UNIX socket HTTP communication.
        
        Parameters:
        - url_scheme: str, URL scheme to mount the adapter for (default: 'http+unix://')
        - *args, **kwargs: Additional arguments passed to requests.Session
        
        Inherits all methods from requests.Session:
        - get(url, **kwargs): Make GET request
        - post(url, data=None, json=None, **kwargs): Make POST request  
        - put(url, data=None, **kwargs): Make PUT request
        - patch(url, data=None, **kwargs): Make PATCH request
        - delete(url, **kwargs): Make DELETE request
        - head(url, **kwargs): Make HEAD request
        - options(url, **kwargs): Make OPTIONS request
        - request(method, url, **kwargs): Make request with specified method
        """

Monkeypatch

Patches the global requests module to support UNIX sockets, either temporarily (as context manager) or globally (as function call).

class monkeypatch:
    def __init__(self, url_scheme=DEFAULT_SCHEME):
        """
        Patches global requests module to support UNIX sockets.
        Can be used as context manager or called directly for global patching.
        
        Parameters:
        - url_scheme: str, URL scheme to handle (default: 'http+unix://')
        
        Usage patterns:
        1. Context manager (temporary): with requests_unixsocket.monkeypatch():
        2. Function call (global): requests_unixsocket.monkeypatch()
        """
    
    def __enter__(self):
        """Enter context manager, patches global requests functions."""
        
    def __exit__(self, *args):
        """Exit context manager, restores original requests functions."""

HTTP Method Functions

Module-level functions that work like their requests counterparts but support UNIX sockets.

def request(method, url, **kwargs):
    """
    Make HTTP request to UNIX socket.
    
    Parameters:
    - method: str, HTTP method ('GET', 'POST', etc.)
    - url: str, UNIX socket URL with http+unix:// scheme
    - **kwargs: Additional arguments passed to requests
    
    Returns:
    requests.Response object
    """

def get(url, **kwargs):
    """
    Make GET request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - **kwargs: Additional arguments (allow_redirects=True by default)
    
    Returns:
    requests.Response object
    """

def post(url, data=None, json=None, **kwargs):
    """
    Make POST request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - data: dict/bytes/file-like, request body data
    - json: dict, JSON data to send in body
    - **kwargs: Additional arguments
    
    Returns:
    requests.Response object
    """

def put(url, data=None, **kwargs):
    """
    Make PUT request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - data: dict/bytes/file-like, request body data
    - **kwargs: Additional arguments
    
    Returns:
    requests.Response object
    """

def patch(url, data=None, **kwargs):
    """
    Make PATCH request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - data: dict/bytes/file-like, request body data
    - **kwargs: Additional arguments
    
    Returns:
    requests.Response object
    """

def delete(url, **kwargs):
    """
    Make DELETE request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - **kwargs: Additional arguments
    
    Returns:
    requests.Response object
    """

def head(url, **kwargs):
    """
    Make HEAD request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - **kwargs: Additional arguments (allow_redirects=False by default)
    
    Returns:
    requests.Response object
    """

def options(url, **kwargs):
    """
    Make OPTIONS request to UNIX socket.
    
    Parameters:
    - url: str, UNIX socket URL
    - **kwargs: Additional arguments (allow_redirects=True by default)
    
    Returns:
    requests.Response object
    """

UnixAdapter

Low-level HTTP adapter for UNIX socket connections.

class UnixAdapter(requests.adapters.HTTPAdapter):
    def __init__(self, timeout=60, pool_connections=25, *args, **kwargs):
        """
        HTTP adapter for UNIX domain socket connections.
        
        Parameters:
        - timeout: int, connection timeout in seconds (default: 60)
        - pool_connections: int, number of connection pools to cache (default: 25)
        - *args, **kwargs: Additional arguments passed to HTTPAdapter
        """
    
    def get_connection(self, url, proxies=None):
        """
        Get connection pool for the given URL.
        
        Parameters:
        - url: str, UNIX socket URL
        - proxies: dict, proxy configuration (not supported, will raise ValueError)
        
        Returns:
        UnixHTTPConnectionPool instance
        
        Raises:
        ValueError: If proxies are specified
        """
    
    def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
        """
        Get connection with TLS context (compatibility method for requests 2.32.2+).
        
        Parameters:
        - request: PreparedRequest object
        - verify: bool/str, SSL verification settings
        - proxies: dict, proxy configuration (not supported)
        - cert: SSL client certificate
        
        Returns:
        Connection pool
        """
    
    def request_url(self, request, proxies):
        """
        Extract path portion from request URL for UNIX socket communication.
        
        Parameters:
        - request: PreparedRequest object
        - proxies: dict, proxy configuration
        
        Returns:
        str: Path portion of the URL
        """
    
    def close(self):
        """Close all connection pools and clean up resources."""

Testing Utilities

UnixSocketServerThread

Test server that creates an HTTP server listening on a UNIX socket.

class UnixSocketServerThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        """
        Creates a test HTTP server on a temporary UNIX socket.
        
        Attributes:
        - usock: str, path to the created UNIX socket
        """
    
    def run(self):
        """Start the server thread (called automatically by threading)."""
    
    def __enter__(self):
        """
        Enter context manager, starts server and waits for it to be ready.
        
        Returns:
        self: The server thread instance
        """
    
    def __exit__(self, *args):
        """Exit context manager, stops and cleans up the server."""

Constants

DEFAULT_SCHEME: str
    # Default URL scheme: 'http+unix://'

Error Handling

The library raises standard requests exceptions:

  • requests.ConnectionError: When unable to connect to the UNIX socket
  • ValueError: When proxies are specified (not supported)
  • requests.exceptions.InvalidSchema: When using unsupported URL schemes

All other requests exceptions (Timeout, HTTPError, etc.) behave normally.

Usage Examples

Docker API Communication

import requests_unixsocket
import json

session = requests_unixsocket.Session()

# Get Docker system information
response = session.get('http+unix://%2Fvar%2Frun%2Fdocker.sock/info')
info = response.json()
print(f"Docker version: {info['ServerVersion']}")

# List containers
response = session.get('http+unix://%2Fvar%2Frun%2Fdocker.sock/containers/json')
containers = response.json()
print(f"Running containers: {len(containers)}")

Testing with UnixSocketServerThread

import requests_unixsocket
from requests_unixsocket.testutils import UnixSocketServerThread

# Create a test server
with UnixSocketServerThread() as server:
    session = requests_unixsocket.Session()
    
    # URL-encode the socket path
    import requests.compat
    socket_path = requests.compat.quote_plus(server.usock)
    url = f'http+unix://{socket_path}/test/endpoint'
    
    # Make request to test server
    response = session.get(url)
    assert response.status_code == 200
    print(response.text)  # "Hello world!"

Global Monkeypatch

import requests
import requests_unixsocket

# Patch globally (affects all requests in the application)
requests_unixsocket.monkeypatch()

# Now regular requests functions work with UNIX sockets
response = requests.get('http+unix://%2Fvar%2Frun%2Fdocker.sock/info')
print(response.status_code)  # 200

# All requests module functions now support UNIX sockets
response = requests.post('http+unix://%2Ftmp%2Fapp.sock/api/data', 
                        json={'key': 'value'})
print(response.json())

Platform Support

  • UNIX domain sockets: Supported on all UNIX-like systems (Linux, macOS, BSD)
  • Abstract namespace sockets: Linux only (use \0 prefix)
  • Python versions: 3.9+