在请求框架中使用 unix 域套接字的正确方法是什么？

Question

通常，使用请求框架执行后请求通过以下方式完成：

payload = {'key1': 'value1', 'key2': 'value2'}
r = requests.post("http://httpbin.org/post", data=payload)

但是：如何连接到 unix 套接字而不是进行 TCP 连接？

在相关说明中，如何在 URL 中对域路径进行编码？

• libcurl 允许应用程序提供自己的套接字来执行请求
• LDAP 发明了自己的方案ldapi其中套接字名称在主机字段中进行 % 编码
• httpie在主机字段中使用http+unix方案和%编码路径

这些是一些示例，但是否有 RFC 或既定的最佳实践？

Answer 1

没有必要重新发明轮子：

https://github.com/msabramo/requests-unixsocket

http+unix URL 方案，套接字路径以百分比编码到主机字段中：

import requests_unixsocket
session = requests_unixsocket.Session()
# Access /path/to/page from /tmp/profilesvc.sock
r = session.get('http+unix://%2Ftmp%2Fprofilesvc.sock/path/to/page')
assert r.status_code == 200

Answer 2

如果你正在寻找一种在Python 3中解决这个问题的简约和干净的方法，这里有一个工作示例，它将与Ubuntu在Unix域套接字上的快照进行对话。

import requests
import socket
import pprint
from urllib3.connection import HTTPConnection
from urllib3.connectionpool import HTTPConnectionPool
from requests.adapters import HTTPAdapter

class SnapdConnection(HTTPConnection):
    def __init__(self):
        super().__init__("localhost")
    def connect(self):
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.connect("/run/snapd.socket")

class SnapdConnectionPool(HTTPConnectionPool):
    def __init__(self):
        super().__init__("localhost")
    def _new_conn(self):
        return SnapdConnection()

class SnapdAdapter(HTTPAdapter):
    def get_connection(self, url, proxies=None):
        return SnapdConnectionPool()

session = requests.Session()
session.mount("http://snapd/", SnapdAdapter())
response = session.get("http://snapd/v2/system-info")
pprint.pprint(response.json())

Answer 3

您可以使用

socat创建TCP到UNIX套接字代理，如下所示：

socat TCP-LISTEN:80,reuseaddr,fork UNIX-CLIENT:/tmp/foo.sock

然后将您的 http 请求发送到该代理。侦听 UNIX 套接字/tmp/foo.sock的服务器仍然必须理解 HTTP，因为socat不进行任何消息转换。

Answer 4

请求

没有实现来使用开箱即用的Unix套接字。

但是您可以创建自定义适配器，该适配器将连接到Unix套接字，发送请求并读取答案。

您需要实现的所有方法都是.send()和.close()，这既简单又直接。

在会话对象中注册适配器后，您可以将请求机制与 UNIX 传输一起使用。

Answer 5

https://stackoverflow.com/users/1105249/luis-masuelli 的另一个解决方案是改用httpx包，

>>> import httpx
>>> # Define a transporter
>>> transport = httpx.HTTPTransport(uds="/tmp/profilesvc.sock")
>>> client = httpx.Client(transport=transport)
>>> response = client.get('http://path/to/page')
>>> assert response.status_code == 200