When they're not builtins, they're boltons.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Socket programming utilities with buffering and protocol support, plus comprehensive URL parsing, manipulation, and utilities. Includes netstring protocol implementation, complete URL component handling, and robust socket wrapper classes for network programming.
Enhanced socket wrappers with buffering and protocol support.
class BufferedSocket:
"""Socket wrapper with buffering capabilities."""
def __init__(self, sock, timeout=DEFAULT_TIMEOUT, maxsize=DEFAULT_MAXSIZE): ...
def recv(self, size, flags=0, timeout=None): ...
def recv_until(self, delimiter, timeout=None, maxsize=None): ...
def recv_size(self, size, timeout=None): ...
def send(self, data, flags=0, timeout=None): ...
def sendall(self, data, timeout=None): ...
def close(self): ...
class NetstringSocket:
"""Socket using netstring protocol for message framing."""
def __init__(self, sock, timeout=DEFAULT_TIMEOUT, maxsize=DEFAULT_MAXSIZE): ...
def read_ns(self, timeout=None, maxsize=None): ...
def write_ns(self, data, timeout=None): ...
def close(self): ...Comprehensive URL handling with immutable semantics.
class URL:
"""Comprehensive URL manipulation class with immutable semantics."""
def __init__(self, url_text='', **kwargs): ...
# Properties
@property
def scheme(self): ...
@property
def username(self): ...
@property
def password(self): ...
@property
def host(self): ...
@property
def port(self): ...
@property
def path(self): ...
@property
def query_params(self): ...
@property
def fragment(self): ...
# Methods
def replace(self, **kwargs): ...
def normalize(self): ...
def __str__(self): ...
def parse_url(url_text):
"""
Parse URL string into components.
Parameters:
- url_text (str): URL string to parse
Returns:
URL: Parsed URL object
"""Utilities for processing individual URL components.
def quote_path_part(text, full_quote=True):
"""
URL-encode path components.
Parameters:
- text (str): Text to encode
- full_quote (bool): Use full URL encoding
Returns:
str: URL-encoded path component
"""
def quote_query_part(text, full_quote=True):
"""
URL-encode query parameters.
Parameters:
- text (str): Text to encode
- full_quote (bool): Use full URL encoding
Returns:
str: URL-encoded query parameter
"""
def quote_fragment_part(text, full_quote=True):
"""
URL-encode fragment components.
Parameters:
- text (str): Text to encode
- full_quote (bool): Use full URL encoding
Returns:
str: URL-encoded fragment
"""
def quote_userinfo_part(text, full_quote=True):
"""
URL-encode userinfo components.
Parameters:
- text (str): Text to encode
- full_quote (bool): Use full URL encoding
Returns:
str: URL-encoded userinfo
"""
def unquote(string, encoding='utf-8', errors='replace'):
"""
URL-decode strings.
Parameters:
- string (str): URL-encoded string
- encoding (str): Character encoding
- errors (str): Error handling strategy
Returns:
str: Decoded string
"""
def unquote_to_bytes(string):
"""
URL-decode to bytes.
Parameters:
- string (str): URL-encoded string
Returns:
bytes: Decoded bytes
"""Extract and analyze URLs from text.
def find_all_links(text, with_text=False, **kwargs):
"""
Extract URLs from text.
Parameters:
- text (str): Text containing URLs
- with_text (bool): Include surrounding text
Returns:
list: List of found URLs or (url, text) tuples
"""
def parse_host(host):
"""
Parse hostname/IP address with port.
Parameters:
- host (str): Host string (hostname:port or IP:port)
Returns:
tuple: (hostname, port) tuple
"""Parse and manipulate URL query strings.
def parse_qsl(qs, keep_blank_values=True, **kwargs):
"""
Parse query string to list.
Parameters:
- qs (str): Query string to parse
- keep_blank_values (bool): Keep parameters with empty values
Returns:
list: List of (key, value) tuples
"""
class QueryParamDict(OrderedMultiDict):
"""Specialized dict for query parameters."""
def add(self, key, value): ...
def get_list(self, key): ...
def to_url_query(self): ...Register and manage custom URL schemes.
def register_scheme(text, uses_netloc=None, **kwargs):
"""
Register custom URL schemes.
Parameters:
- text (str): Scheme name
- uses_netloc (bool): Whether scheme uses network location
Returns:
None
"""
def resolve_path_parts(path_parts):
"""
Resolve relative path components.
Parameters:
- path_parts (list): List of path components
Returns:
list: Resolved path components
"""from boltons.socketutils import BufferedSocket, NetstringSocket\nfrom boltons.urlutils import URL, find_all_links, parse_qsl\nimport socket\n\n# Enhanced socket operations\nsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\nbuf_sock = BufferedSocket(sock)\nbuf_sock.sendall(b'HTTP/1.1 request data')\nresponse = buf_sock.recv_until(b'\\r\\n\\r\\n') # Read until headers end\n\n# Netstring protocol\nns_sock = NetstringSocket(sock)\nns_sock.write_ns(b'structured message')\nmessage = ns_sock.read_ns()\n\n# URL manipulation\nurl = URL('https://example.com/path?param=value#section')\nprint(url.host) # 'example.com'\nprint(url.path) # '/path'\nprint(url.query_params) # QueryParamDict([('param', 'value')])\n\n# Immutable URL modifications\nnew_url = url.replace(path='/new-path', query_params={'new': 'param'})\nprint(new_url) # 'https://example.com/new-path?new=param#section'\n\n# Extract URLs from text\ntext = \"Visit https://example.com and http://test.org for more info\"\nurls = find_all_links(text)\nprint(urls) # ['https://example.com', 'http://test.org']\n\n# Query string processing\nquery = \"param1=value1¶m2=value2¶m1=another_value\"\nparams = parse_qsl(query)\nprint(params) # [('param1', 'value1'), ('param2', 'value2'), ('param1', 'another_value')]\n```
### Advanced URL Operations
```python
from boltons.urlutils import (\n URL, quote_path_part, unquote, register_scheme\n)\n\n# URL component encoding\npath = \"path with spaces/special chars!\"\nencoded = quote_path_part(path)\nprint(encoded) # \"path%20with%20spaces/special%20chars%21\"\n\n# URL decoding\nencoded_url = \"https%3A//example.com/path%20with%20spaces\"\ndecoded = unquote(encoded_url)\nprint(decoded) # \"https://example.com/path with spaces\"\n\n# Complex URL building\nbase_url = URL('https://api.example.com')\napi_url = base_url.replace(\n path='/v1/users',\n query_params={\n 'limit': 10,\n 'offset': 20,\n 'filter': ['active', 'verified']\n }\n)\nprint(api_url)\n# https://api.example.com/v1/users?limit=10&offset=20&filter=active&filter=verified\n\n# Custom scheme registration\nregister_scheme('myprotocol', uses_netloc=True)\ncustom_url = URL('myprotocol://server:port/path')\nprint(custom_url.scheme) # 'myprotocol'\nprint(custom_url.host) # 'server'\nprint(custom_url.port) # port\n```
## Types
```python { .api }
# Socket Exceptions\nclass Error(socket.error):\n \"\"\"Base exception for socket utilities.\"\"\"\n pass\n\nclass ConnectionClosed(Error):\n \"\"\"Exception for closed connections.\"\"\"\n pass\n\nclass MessageTooLong(Error):\n \"\"\"Exception for oversized messages.\"\"\"\n pass\n\nclass Timeout(socket.timeout, Error):\n \"\"\"Timeout exception.\"\"\"\n pass\n\nclass NetstringProtocolError(Error):\n \"\"\"Base netstring protocol error.\"\"\"\n pass\n\nclass NetstringInvalidSize(NetstringProtocolError):\n \"\"\"Invalid netstring size error.\"\"\"\n pass\n\nclass NetstringMessageTooLong(NetstringProtocolError):\n \"\"\"Netstring message too long error.\"\"\"\n pass\n\n# URL Exceptions\nclass URLParseError(ValueError):\n \"\"\"Exception for URL parsing errors.\"\"\"\n pass\n\n# Constants\nDEFAULT_TIMEOUT = 10 # Default timeout in seconds\nDEFAULT_MAXSIZE = 32 * 1024 # Default maximum message size (32KB)\nDEFAULT_ENCODING = 'utf8' # Default character encoding\n\n# URL scheme mappings\nSCHEME_PORT_MAP: dict # Mapping of URL schemes to default ports\nNO_NETLOC_SCHEMES: set # Set of schemes that don't use netloc\nDEFAULT_PARSED_URL: dict # Default empty parsed URL structure\n```Install with Tessl CLI
npx tessl i tessl/pypi-boltons