Connection adapter for Requests that allows it to talk with raw UNIX sockets.

socketadapter.py

"""
Connection adapter for Requests that allows it to talk with raw UNIX sockets.

Adapted from requests-unixsocket, which itself was adapted from docker-py.
https://github.com/msabramo/requests-unixsocket
https://github.com/docker/docker-py/blob/master/docker/unixconn/unixconn.py
"""

import socket
from urllib.parse import unquote, urlparse
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.connection import HTTPConnection
from requests.packages.urllib3.connectionpool import HTTPConnectionPool


class UnixHTTPConnection(HTTPConnection):
    def __init__(self, unix_socket_url, timeout=60):
        """
        Creates an HTTP connection to a UNIX domain socket.

        The URL scheme is 'http+unix' and the netloc is a percent-encoded path
        to a UNIX domain socket, e.g.
        'http+unix://%2Fvar%2Frun%2Fdocker.sock/containers/json'
        """
        super().__init__('localhost', timeout=timeout)
        self.unix_socket_url = unix_socket_url
        self.timeout = timeout

    def connect(self):
        socket_path = unquote(urlparse(self.unix_socket_url).netloc)
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.settimeout(self.timeout)
        sock.connect(socket_path)
        self.sock = sock


class UnixHTTPConnectionPool(HTTPConnectionPool):
    def __init__(self, socket_path, timeout=60):
        super().__init__('localhost', timeout=timeout)
        self.socket_path = socket_path
        self.timeout = timeout

    def _new_conn(self):
        return UnixHTTPConnection(self.socket_path, self.timeout)


class UnixAdapter(HTTPAdapter):
    def __init__(self, timeout=60):
        super().__init__()
        self.timeout = timeout

    def get_connection(self, socket_path, proxies=None):
        if proxies:
            raise ValueError('{} lacks support for proxies.'.format(
                self.__class__.__name__))

        return UnixHTTPConnectionPool(socket_path, self.timeout)

I use this to talk to the Docker Remote API, though it should be useful for other HTTP APIs available over a UNIX socket.