Utilities

def make_headers(basic_auth=None, proxy_basic_auth=None, user_agent=None,
                keep_alive=None, accept_encoding=None, disable_cache=None) -> dict:
    """
    Generate common HTTP headers.

    Parameters:
    - basic_auth: Username and password for HTTP Basic Auth (tuple or string)
    - proxy_basic_auth: Username and password for proxy Basic Auth (tuple or string)  
    - user_agent: User-Agent header value
    - keep_alive: Whether to include Keep-Alive header (bool)
    - accept_encoding: Accept-Encoding header value (bool or list)
    - disable_cache: Whether to include cache-disabling headers (bool)

    Returns:
    dict: Dictionary of HTTP headers
    """

def encode_multipart_formdata(fields, boundary=None) -> tuple[bytes, str]:
    """
    Encode multipart form data for HTTP requests.

    Parameters:
    - fields: Sequence of (name, value) tuples or RequestField objects
    - boundary: Multipart boundary string (auto-generated if None)

    Returns:
    tuple[bytes, str]: Encoded form data and content-type header value
    """

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

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

    Returns:
    HTTPConnectionPool: Configured connection pool for the URL's host
    """

def proxy_from_url(url: str, **kw) -> ProxyManager:
    """
    Create ProxyManager from proxy URL.

    Parameters:
    - url: Proxy URL (e.g., 'http://proxy.example.com:8080')
    - **kw: Additional arguments for ProxyManager

    Returns:
    ProxyManager: Configured proxy manager
    """

def parse_url(url: str) -> Url:
    """
    Parse URL string into components.

    Parameters:
    - url: URL string to parse

    Returns:
    Url: Named tuple with scheme, auth, host, port, path, query, fragment
    """

class Url:
    """URL components as named tuple"""
    scheme: str      # http, https, etc.
    auth: str        # username:password
    host: str        # hostname or IP
    port: int        # port number
    path: str        # path component
    query: str       # query string
    fragment: str    # fragment identifier

def create_urllib3_context(ssl_version=None, cert_reqs=None, options=None,
                          ciphers=None) -> ssl.SSLContext:
    """
    Create SSL context with urllib3 defaults.

    Parameters:
    - ssl_version: SSL/TLS version to use
    - cert_reqs: Certificate requirements (ssl.CERT_NONE, ssl.CERT_REQUIRED)
    - options: SSL context options
    - ciphers: Cipher suite string

    Returns:
    ssl.SSLContext: Configured SSL context
    """

def ssl_wrap_socket(sock, keyfile=None, certfile=None, cert_reqs=None,
                   ca_certs=None, server_hostname=None, ssl_version=None,
                   ciphers=None, ssl_context=None) -> ssl.SSLSocket:
    """
    Wrap socket with SSL/TLS.

    Returns:
    ssl.SSLSocket: SSL-wrapped socket
    """

def assert_fingerprint(cert, fingerprint):
    """
    Verify SSL certificate fingerprint.

    Parameters:
    - cert: Certificate in DER format
    - fingerprint: Expected fingerprint (hex string)

    Raises:
    SSLError: If fingerprint doesn't match
    """

def resolve_cert_reqs(candidate) -> int:
    """
    Resolve certificate requirements to ssl module constant.

    Parameters:
    - candidate: String ('CERT_NONE', 'CERT_REQUIRED') or ssl constant

    Returns:
    int: ssl.CERT_NONE or ssl.CERT_REQUIRED
    """

def resolve_ssl_version(candidate) -> int:
    """
    Resolve SSL version to ssl module constant.

    Parameters:
    - candidate: String ('TLSv1', 'TLSv1_2') or ssl constant

    Returns:
    int: SSL version constant
    """

def is_connection_dropped(conn) -> bool:
    """
    Check if connection has been dropped by the server.

    Parameters:
    - conn: Connection object to check

    Returns:
    bool: True if connection appears to be dropped
    """

def wait_for_read(sock, timeout=None) -> bool:
    """
    Wait for socket to be ready for reading.

    Parameters:
    - sock: Socket object
    - timeout: Timeout in seconds (None for no timeout)

    Returns:
    bool: True if socket is ready for reading
    """

def wait_for_write(sock, timeout=None) -> bool:
    """
    Wait for socket to be ready for writing.

    Parameters:
    - sock: Socket object  
    - timeout: Timeout in seconds (None for no timeout)

    Returns:
    bool: True if socket is ready for writing
    """

class RequestField:
    def __init__(self, name, data, filename=None, headers=None):
        """
        Represents a single multipart form field.

        Parameters:
        - name: Field name
        - data: Field data (string, bytes, or file-like object)
        - filename: Filename for file fields
        - headers: Additional headers for this field
        """

    def make_multipart(self, content_disposition=None, content_type=None,
                      content_location=None) -> RequestField:
        """Configure field for multipart encoding"""

def guess_content_type(filename, default='application/octet-stream') -> str:
    """
    Guess MIME content type from filename.

    Parameters:
    - filename: Filename to analyze
    - default: Default content type if guess fails

    Returns:
    str: MIME content type
    """

import urllib3

# Basic authentication headers
headers = urllib3.make_headers(basic_auth='username:password')
print(headers)  # {'authorization': 'Basic dXNlcm5hbWU6cGFzc3dvcmQ='}

# User agent header
headers = urllib3.make_headers(user_agent='MyApp/1.0')
print(headers)  # {'user-agent': 'MyApp/1.0'}

# Multiple headers
headers = urllib3.make_headers(
    basic_auth=('user', 'pass'),
    user_agent='MyApp/1.0',
    keep_alive=True,
    accept_encoding=True
)

http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/headers', headers=headers)

import urllib3

# Simple form fields
fields = [
    ('username', 'john'),
    ('password', 'secret'),
    ('remember', 'on')
]

body, content_type = urllib3.encode_multipart_formdata(fields)

http = urllib3.PoolManager()
resp = http.request('POST', 'https://httpbin.org/post',
                   body=body,
                   headers={'Content-Type': content_type})

import urllib3

# Upload file with form data
with open('/path/to/file.txt', 'rb') as f:
    fields = [
        ('file', (f.name, f.read(), 'text/plain')),
        ('description', 'My uploaded file'),
        ('category', 'documents')
    ]

    body, content_type = urllib3.encode_multipart_formdata(fields)

http = urllib3.PoolManager()
resp = http.request('POST', 'https://httpbin.org/post',
                   body=body,
                   headers={'Content-Type': content_type})

import urllib3

# Create custom request field
field = urllib3.fields.RequestField(
    name='document',
    data=open('/path/to/document.pdf', 'rb').read(),
    filename='document.pdf'
)
field.make_multipart(content_type='application/pdf')

# Combine with other fields
fields = [
    field,
    ('title', 'Important Document'),
    ('tags', 'important,pdf')
]

body, content_type = urllib3.encode_multipart_formdata(fields)

http = urllib3.PoolManager()
resp = http.request('POST', 'https://api.example.com/upload',
                   body=body,
                   headers={'Content-Type': content_type})

import urllib3

# Parse complex URL
url = urllib3.util.parse_url('https://user:pass@api.example.com:8443/v1/data?format=json#results')

print(f"Scheme: {url.scheme}")      # https
print(f"Auth: {url.auth}")          # user:pass  
print(f"Host: {url.host}")          # api.example.com
print(f"Port: {url.port}")          # 8443
print(f"Path: {url.path}")          # /v1/data
print(f"Query: {url.query}")        # format=json
print(f"Fragment: {url.fragment}")  # results

# Create connection pool from parsed URL
pool = urllib3.connection_from_url(str(url))

import urllib3
import ssl

# Create custom SSL context
ssl_context = urllib3.util.create_urllib3_context(
    ssl_version=ssl.PROTOCOL_TLS,
    cert_reqs=ssl.CERT_REQUIRED,
    ciphers='HIGH:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!SRP:!CAMELLIA'
)

# Use with connection pool
pool = urllib3.HTTPSConnectionPool(
    'secure.example.com',
    ssl_context=ssl_context
)

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

import urllib3

def verify_fingerprint_callback(conn, cert, errno, depth, ok):
    """Custom certificate verification with fingerprint check"""
    if depth == 0:  # Only check end-entity certificate
        try:
            urllib3.util.assert_fingerprint(cert, 'AA:BB:CC:DD:...')
            return True
        except urllib3.exceptions.SSLError:
            return False
    return ok

# This would be used in a custom SSL context setup

import urllib3
import socket

# Test if connection is still alive
http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/get')

# Get underlying connection from response
conn = resp.connection

# Check if connection was dropped
if urllib3.util.is_connection_dropped(conn):
    print("Connection was dropped by server")
else:
    print("Connection is still alive")

resp.close()

def is_fp_closed(obj) -> bool:
    """
    Check if file-like object is closed.

    Parameters:
    - obj: File-like object to check

    Returns:
    bool: True if object is closed or unavailable
    """

import urllib3

# Create proxy manager from URL
proxy = urllib3.proxy_from_url('http://proxy.example.com:8080')

# Or with authentication
proxy = urllib3.proxy_from_url('http://user:pass@proxy.example.com:8080')

# Use proxy for requests
resp = proxy.request('GET', 'https://httpbin.org/ip')
print(f"IP through proxy: {resp.json()}")

import urllib3

# Generate headers for different scenarios
auth_headers = urllib3.make_headers(basic_auth='admin:secret')
proxy_headers = urllib3.make_headers(proxy_basic_auth='proxy_user:proxy_pass')
ua_headers = urllib3.make_headers(user_agent='Mozilla/5.0 (Custom Bot)')

# Combine multiple header sets
combined_headers = {}
combined_headers.update(auth_headers)
combined_headers.update(ua_headers)
combined_headers.update({'Custom-Header': 'CustomValue'})

http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/headers',
                   headers=combined_headers)

import urllib3

# Guess content type from filename
content_type = urllib3.fields.guess_content_type('document.pdf')
print(content_type)  # application/pdf

content_type = urllib3.fields.guess_content_type('image.jpg') 
print(content_type)  # image/jpeg

content_type = urllib3.fields.guess_content_type('unknown.xyz')
print(content_type)  # application/octet-stream

def add_stderr_logger(level: int = logging.DEBUG) -> logging.StreamHandler:
    """
    Add a StreamHandler to the logger for debugging purposes.

    Parameters:
    - level: Logging level (default: logging.DEBUG)

    Returns:
    logging.StreamHandler: The added handler
    """

def disable_warnings(category: type[Warning] = exceptions.HTTPWarning) -> None:
    """
    Helper for quickly disabling urllib3 warnings.

    Parameters:
    - category: Warning category to disable (default: HTTPWarning)
    """

import urllib3
import logging

# Enable debug logging
handler = urllib3.add_stderr_logger(level=logging.DEBUG)

# Make a request with debug output
http = urllib3.PoolManager()
resp = http.request('GET', 'https://httpbin.org/get')

# Disable specific warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Disable all urllib3 warnings
urllib3.disable_warnings()

# Make request that would normally generate warnings
resp = http.request('GET', 'https://self-signed.badssl.com/', verify=False)

def format_multipart_header_param(name: str, value: str | bytes) -> str:
    """
    Format and quote a single multipart header parameter.

    Parameters:
    - name: Parameter name (ASCII-only string)
    - value: Parameter value (string or UTF-8 encoded bytes)

    Returns:
    str: Formatted parameter string 'name="value"' with proper escaping
    """

class RequestField:
    @classmethod
    def from_tuples(cls, fieldname: str, value, header_formatter=None) -> RequestField:
        """
        Create RequestField from old-style tuple parameters.

        Parameters:
        - fieldname: Field name
        - value: Field value (string, bytes, or tuple of (filename, data, mime_type))
        - header_formatter: Custom header formatter function

        Returns:
        RequestField: Configured request field
        """

import urllib3

# Create RequestField from tuple (filename, data, mime_type)
field = urllib3.fields.RequestField.from_tuples(
    'document',
    ('report.pdf', open('report.pdf', 'rb').read(), 'application/pdf')
)

# Create RequestField from simple value
field = urllib3.fields.RequestField.from_tuples('username', 'john_doe')

# Custom multipart parameter formatting
formatted = urllib3.fields.format_multipart_header_param(
    'filename', 'my document.txt'
)
print(formatted)  # filename="my%20document.txt"

# Use in multipart form data
fields = [
    urllib3.fields.RequestField.from_tuples('file', ('data.csv', csv_data, 'text/csv')),
    urllib3.fields.RequestField.from_tuples('description', 'Monthly report')
]

body, content_type = urllib3.encode_multipart_formdata(fields)

# Skip header generation (sentinel value)
SKIP_HEADER = object()

# Headers that can be skipped in make_headers
SKIPPABLE_HEADERS = frozenset([
    'accept-encoding', 'host', 'user-agent'
])

# SSL/TLS availability
IS_PYOPENSSL = bool(...)  # True if PyOpenSSL is available
HAS_SNI = bool(...)       # True if SNI support is available

# Default SSL protocols
ALPN_PROTOCOLS = ['h2', 'http/1.1']  # Supported ALPN protocols