requests-unixsocket2
requests-unixsocket2 allows the popular 'requests' HTTP library to communicate over UNIX domain sockets. It is a maintained fork of the original 'requests-unixsocket' project, created to ensure compatibility with recent versions of 'requests' and 'urllib3'. The current version is 1.0.1, with releases typically occurring as needed to address compatibility issues or bug fixes.
Warnings
- breaking The original `requests-unixsocket` library is considered abandoned and is incompatible with recent versions of `requests` and `urllib3`. `requests-unixsocket2` was created to address these compatibility issues and is the recommended alternative.
- gotcha Direct calls to `requests.get()`, `requests.post()`, etc., will not utilize UNIX domain sockets by default. You must either instantiate `requests_unixsocket.Session()` and use its methods, or call `requests_unixsocket.monkeypatch()` to globally enable UNIX socket support for `requests` functions.
- gotcha Abstract namespace sockets (e.g., URLs like `http+unix://\0test_socket/path`) are a Linux-specific feature. Code using abstract namespace sockets will not work on other operating systems.
Install
-
pip install requests-unixsocket2
Imports
- Session
from requests_unixsocket import Session
- monkeypatch
from requests_unixsocket import monkeypatch
Quickstart
import requests_unixsocket
import json
import os
# Example: Connect to a hypothetical service via a UNIX domain socket.
# Replace '/tmp/my_service.sock' with your actual socket path and '/info' with your API endpoint.
# The socket path must be URL-encoded, e.g., '/' becomes '%2F'.
# For demonstration purposes, use an environment variable or a common placeholder.
# In a real application, ensure the service is running and listening on this socket.
unix_socket_path = os.environ.get('UNIX_SOCKET_PATH', '/tmp/my_service.sock')
encoded_unix_socket_path = unix_socket_path.replace("/", "%2F")
session = requests_unixsocket.Session()
try:
# Construct the URL using the 'http+unix://' scheme
response = session.get(f'http+unix://{encoded_unix_socket_path}/info')
response.raise_for_status() # Raises an HTTPError for bad responses (4xx or 5xx)
print("Response from UNIX socket service:")
print(json.dumps(response.json(), indent=2))
except requests_unixsocket.exceptions.ConnectionError as e:
print(f"Could not connect to UNIX socket at {unix_socket_path}: {e}")
print("Please ensure a service is running and listening on this socket.")
except json.JSONDecodeError:
print(f"Received non-JSON response or empty response from {unix_socket_path}. Status: {response.status_code}")
print(f"Response text: {response.text}")
except Exception as e:
print(f"An unexpected error occurred: {e}")