Python Requests：如何在发送请求前获取客户端端口？

方案一：预创建Socket并手动绑定（最可靠） #

自己提前创建Socket，获取本地端口后再交给urllib3使用，无论请求成功或失败，都能提前记录端口：

import socket
import urllib3
from urllib3.connection import HTTPConnection

def create_tracked_socket(host, port):
    # 创建Socket并让系统分配随机端口
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(('', 0))
    local_port = sock.getsockname()[1]
    # 可将端口与请求上下文绑定存储，比如存入全局/进程本地容器
    return sock, local_port

# 自定义Connection类，替换默认socket创建逻辑
class TrackedHTTPConnection(HTTPConnection):
    def __init__(self, *args, **kwargs):
        self.tracked_port = None
        super().__init__(*args, **kwargs)
    
    def connect(self):
        self.sock, self.tracked_port = create_tracked_socket(self.host, self.port)
        self.sock.connect((self.host, self.port))

# 使用自定义Connection初始化连接池
manager = urllib3.PoolManager(connection_class=TrackedHTTPConnection)

try:
    resp = manager.request('GET', 'http://example.com')
except urllib3.exceptions.ConnectionError as e:
    # 从异常关联的connection实例中提取端口
    conn = e.__context__.connection
    if conn and hasattr(conn, 'tracked_port'):
        print(f"请求失败，本地端口: {conn.tracked_port}")

方案二：Monkey Patch urllib3的Socket创建函数 #

urllib3内部通过urllib3.util.connection.create_connection创建Socket，我们可以包装这个函数，在返回Socket前记录本地端口：

import urllib3
from urllib3.util.connection import create_connection as original_create_connection
from multiprocessing import current_process, Manager

# 多进程环境下用共享字典存储端口关联关系
port_tracker = Manager().dict()

def tracked_create_connection(address, *args, **kwargs):
    sock = original_create_connection(address, *args, **kwargs)
    local_port = sock.getsockname()[1]
    # 用进程ID+目标地址作为键，存储对应本地端口
    pid = current_process().pid
    port_tracker[(pid, address)] = local_port
    return sock

# 替换原创建函数
urllib3.util.connection.create_connection = tracked_create_connection

# 正常使用urllib3连接池
manager = urllib3.PoolManager()

try:
    resp = manager.request('GET', 'http://example.com')
except urllib3.exceptions.ConnectionError as e:
    target_address = (e.__context__.connection.host, e.__context__.connection.port)
    pid = current_process().pid
    failed_port = port_tracker.get((pid, target_address))
    if failed_port:
        print(f"请求失败，本地端口: {failed_port}")

方案三：利用urllib3的连接创建钩子 #

部分urllib3版本支持connection_created事件回调，可在Socket创建后触发端口记录逻辑：

import urllib3
from urllib3.connectionpool import HTTPConnectionPool

def track_port(conn):
    if hasattr(conn, 'sock'):
        local_port = conn.sock.getsockname()[1]
        conn.tracked_port = local_port  # 将端口绑定到connection实例

# 注册钩子
HTTPConnectionPool.connection_created = track_port

manager = urllib3.PoolManager()

try:
    resp = manager.request('GET', 'http://example.com')
except urllib3.exceptions.ConnectionError as e:
    conn = e.__context__.connection
    if conn and hasattr(conn, 'tracked_port'):
        print(f"请求失败，本地端口: {conn.tracked_port}")

注意事项 #

• 多进程环境下，端口存储要注意进程隔离，优先用进程本地存储或multiprocessing.Manager提供的共享容器。
• 异常处理时，不同ConnectionError子类的属性结构可能不同，可临时打印异常的__dict__来排查关联的connection实例位置。
• 若请求HTTPS站点，需替换为继承urllib3.connection.HTTPSConnection，核心逻辑保持一致。

内容的提问来源于stack exchange，提问作者Jules Olléon