Connection Pools

class HTTPConnectionPool:
    def __init__(self, host: str, port=None, timeout=None, maxsize=1,
                block=False, headers=None, retries=None, **conn_kw):
        """
        HTTP connection pool for a specific host.

        Parameters:
        - host: Host to connect to
        - port: Port number (default: 80 for HTTP, 443 for HTTPS)
        - timeout: Default timeout for requests (float, Timeout object, or None)
        - maxsize: Maximum number of connections in the pool
        - block: Block when pool is full instead of raising FullPoolError
        - headers: Default headers for all requests
        - retries: Default retry configuration (Retry object or None)
        - **conn_kw: Additional connection arguments
        """

    def urlopen(self, method: str, url: str, **kw) -> BaseHTTPResponse:
        """
        Make a request using a connection from the pool.

        Parameters:
        - method: HTTP method (GET, POST, etc.)
        - url: Path portion of URL (not including host)
        - **kw: Additional request arguments

        Returns:
        BaseHTTPResponse: Response object
        """

    def close(self):
        """
        Close all pooled connections and disable the pool.
        """

    def is_same_host(self, url: str) -> bool:
        """
        Check if URL is for the same host as this pool.
        """

class HTTPSConnectionPool(HTTPConnectionPool):
    def __init__(self, host: str, port=None, timeout=None, maxsize=1,
                block=False, headers=None, retries=None, **conn_kw):
        """
        HTTPS connection pool with SSL/TLS support.

        Additional SSL-specific parameters in conn_kw:
        - ssl_context: Custom SSL context
        - cert_file: Path to client certificate file
        - key_file: Path to client private key file
        - cert_reqs: Certificate requirements (ssl.CERT_NONE, ssl.CERT_REQUIRED)
        - ca_certs: Path to CA certificate bundle
        - ssl_version: SSL/TLS version to use
        - assert_hostname: Hostname to verify in certificate
        - assert_fingerprint: Expected certificate fingerprint
        """

def connection_from_url(url: str, **kw) -> HTTPConnectionPool:
    """
    Create an appropriate connection pool from a URL.

    Parameters:
    - url: Complete URL (scheme determines HTTP vs HTTPS pool)
    - **kw: Additional arguments for connection pool

    Returns:
    HTTPConnectionPool or HTTPSConnectionPool: Configured connection pool
    """

import urllib3

# Create HTTP connection pool
pool = urllib3.HTTPConnectionPool('httpbin.org', port=80)

# Make requests using the pool
resp1 = pool.urlopen('GET', '/get')
resp2 = pool.urlopen('POST', '/post', fields={'key': 'value'})

print(f"GET response: {resp1.status}")
print(f"POST response: {resp2.status}")

# Close the pool when done
pool.close()

import urllib3

# Create HTTPS connection pool
pool = urllib3.HTTPSConnectionPool('httpbin.org', port=443)

# Make HTTPS requests
resp = pool.urlopen('GET', '/get')
print(f"HTTPS response: {resp.status}")

pool.close()

import urllib3

# Pool with custom settings
pool = urllib3.HTTPSConnectionPool(
    'api.example.com',
    port=443,
    maxsize=10,           # Up to 10 concurrent connections
    block=True,           # Block when pool is full
    timeout=urllib3.Timeout(connect=2.0, read=5.0),
    retries=urllib3.Retry(total=3, backoff_factor=0.3),
    headers={'User-Agent': 'MyApp/1.0'}
)

# Make requests with the configured defaults
resp = pool.urlopen('GET', '/api/data')

import urllib3
import ssl

# Custom SSL context
ctx = ssl.create_default_context()
ctx.check_hostname = True
ctx.verify_mode = ssl.CERT_REQUIRED

# HTTPS pool with custom SSL context
pool = urllib3.HTTPSConnectionPool(
    'secure.example.com',
    ssl_context=ctx
)

resp = pool.urlopen('GET', '/secure-endpoint')

import urllib3

# HTTPS pool with client certificate
pool = urllib3.HTTPSConnectionPool(
    'client-cert.example.com',
    cert_file='/path/to/client.crt',
    key_file='/path/to/client.key',
    ca_certs='/path/to/ca-bundle.crt'
)

resp = pool.urlopen('GET', '/authenticated-endpoint')

import urllib3

# Create pool from URL
pool = urllib3.connection_from_url('https://api.example.com:8443')

# Use the pool
resp = pool.urlopen('GET', '/api/v1/data')
print(f"Response: {resp.status}")

pool.close()

import urllib3

# Context manager usage (recommended)
with urllib3.HTTPSConnectionPool('api.example.com') as pool:
    resp1 = pool.urlopen('GET', '/endpoint1')
    resp2 = pool.urlopen('GET', '/endpoint2')
    # Pool is automatically closed when exiting the context

# Manual lifecycle management
pool = urllib3.HTTPConnectionPool('api.example.com')
try:
    resp = pool.urlopen('GET', '/data')
    # Process response
finally:
    pool.close()  # Always close the pool

import urllib3

# Pool with comprehensive configuration
pool = urllib3.HTTPSConnectionPool(
    'api.example.com',
    
    # Connection pool settings
    maxsize=20,                    # Max connections in pool
    block=True,                    # Block when pool full
    
    # Timeout configuration
    timeout=urllib3.Timeout(
        connect=2.0,               # Connection timeout
        read=10.0                  # Read timeout
    ),
    
    # Retry configuration
    retries=urllib3.Retry(
        total=3,                   # Total retry attempts
        connect=2,                 # Connection retry attempts
        read=2,                    # Read retry attempts
        backoff_factor=0.3         # Exponential backoff
    ),
    
    # SSL/TLS settings
    cert_reqs='CERT_REQUIRED',     # Require valid certificate
    ca_certs='/etc/ssl/certs/ca-certificates.crt',
    
    # Default headers
    headers={
        'User-Agent': 'MyApp/2.0',
        'Accept': 'application/json'
    }
)

# All requests use these default settings
resp = pool.urlopen('GET', '/api/data')