tessl/pypi-tornado
Tornado is a Python web framework and asynchronous networking library designed for applications requiring long-lived connections to many users.
utilities.mddocs/
Utilities and Configuration
Utility functions, configuration management, logging, localization, process management, and string processing utilities that support the core framework functionality.
Capabilities
Configuration and Options
Command-line and configuration file parsing with type validation and automatic help generation.
def define(name: str, default=None, type_=None, help: str = None, metavar: str = None, multiple: bool = False, group: str = None, callback=None):
"""
Define command-line option in global namespace.
Args:
name: Option name
default: Default value
type_: Option type (str, int, float, bool, datetime)
help: Help text
metavar: Metavar for help
multiple: Whether option can be specified multiple times
group: Option group name
callback: Callback when option is parsed
"""
def parse_command_line(args=None, final: bool = True):
"""
Parse global options from command line.
Args:
args: Command line arguments (default: sys.argv)
final: Whether this is final parse
"""
def parse_config_file(path: str, final: bool = True):
"""
Parse global options from config file.
Args:
path: Config file path
final: Whether this is final parse
"""
def print_help(file=None):
"""
Print command line options help.
Args:
file: Output file (default: sys.stdout)
"""
def add_parse_callback(callback):
"""
Add callback to be called after parsing.
Args:
callback: Callback function
"""
class OptionParser:
"""Collection of options with object-like access."""
def __init__(self):
"""Initialize option parser."""
def define(self, name: str, default=None, type_=None, help: str = None, metavar: str = None, multiple: bool = False, group: str = None, callback=None):
"""Define option in this parser."""
def parse_command_line(self, args=None, final: bool = True):
"""Parse command line arguments."""
def parse_config_file(self, path: str, final: bool = True):
"""Parse configuration file."""
def print_help(self, file=None):
"""Print help message."""
def add_parse_callback(self, callback):
"""Add parse callback."""
def mockable(self):
"""Return mockable version for testing."""
# Global options instance
options = OptionParser()String Escaping and Encoding
String processing utilities for HTML/XML escaping, URL encoding, JSON handling, and text formatting.
def xhtml_escape(value) -> str:
"""
Escape string for HTML/XML output.
Args:
value: Value to escape
Returns:
Escaped string
"""
def xhtml_unescape(value: str) -> str:
"""
Unescape XML-escaped string.
Args:
value: Escaped string
Returns:
Unescaped string
"""
def url_escape(value: str, plus: bool = True) -> str:
"""
URL-encode value.
Args:
value: Value to encode
plus: Whether to use + for spaces
Returns:
URL-encoded string
"""
def url_unescape(value: str, encoding: str = 'utf-8', plus: bool = True) -> str:
"""
Decode URL-encoded value.
Args:
value: URL-encoded string
encoding: Character encoding
plus: Whether + represents spaces
Returns:
Decoded string
"""
def json_encode(value) -> str:
"""
JSON-encode Python object.
Args:
value: Object to encode
Returns:
JSON string
"""
def json_decode(value: str):
"""
Decode JSON string to Python object.
Args:
value: JSON string
Returns:
Decoded Python object
"""
def squeeze(value: str) -> str:
"""
Replace sequences of whitespace with single space.
Args:
value: String to process
Returns:
Squeezed string
"""
def utf8(value) -> bytes:
"""
Convert string to UTF-8 bytes.
Args:
value: String or bytes
Returns:
UTF-8 encoded bytes
"""
def to_unicode(value) -> str:
"""
Convert value to unicode string.
Args:
value: Value to convert
Returns:
Unicode string
"""
def recursive_unicode(obj):
"""
Convert data structure to unicode recursively.
Args:
obj: Object to convert
Returns:
Unicode version of object
"""
def linkify(text: str, shorten: bool = False, extra_params: str = "", require_protocol: bool = False, permitted_protocols: List[str] = None) -> str:
"""
Convert URLs in text to HTML links.
Args:
text: Text to process
shorten: Whether to shorten long URLs
extra_params: Extra link parameters
require_protocol: Whether to require protocol
permitted_protocols: Allowed protocols
Returns:
Text with HTML links
"""
def parse_qs_bytes(qs: bytes, keep_blank_values: bool = False, strict_parsing: bool = False) -> Dict[str, List[bytes]]:
"""
Parse query string as bytes.
Args:
qs: Query string bytes
keep_blank_values: Keep blank values
strict_parsing: Strict parsing mode
Returns:
Dict of parameter lists
"""Logging Utilities
Enhanced logging support with colored output and Tornado-specific formatting.
def enable_pretty_logging(options=None, logger=None):
"""
Enable colored console logging.
Args:
options: Options object with logging settings
logger: Logger to configure (default: root logger)
"""
def define_logging_options(options=None):
"""
Define logging command-line options.
Args:
options: OptionParser to add options to
"""
class LogFormatter(logging.Formatter):
"""Log formatter with color support."""
def __init__(self, fmt: str = None, datefmt: str = None, style: str = '%', color: bool = True, colors: Dict[int, int] = None):
"""
Initialize formatter.
Args:
fmt: Log format string
datefmt: Date format string
style: Format style
color: Enable color output
colors: Color mapping for log levels
"""
def format(self, record) -> str:
"""Format log record."""Process Management
Multi-process utilities for forking processes and managing subprocesses.
def cpu_count() -> int:
"""
Get number of processors.
Returns:
Number of CPU cores
"""
def fork_processes(num_processes: int, max_restarts: int = 100) -> int:
"""
Start multiple worker processes.
Args:
num_processes: Number of processes to fork
max_restarts: Maximum restart attempts
Returns:
Process ID (0 for children, None for parent)
"""
def task_id() -> int:
"""
Get current task ID (process number).
Returns:
Task ID starting from 0
"""
class Subprocess:
"""Async subprocess wrapper."""
def __init__(self, *args, **kwargs):
"""
Initialize subprocess.
Args:
*args: subprocess.Popen arguments
**kwargs: subprocess.Popen keyword arguments
"""
def set_exit_callback(self, callback):
"""
Set callback for process exit.
Args:
callback: Exit callback function
"""
async def wait_for_exit(self, raise_error: bool = True):
"""
Wait for process to exit.
Args:
raise_error: Raise exception on non-zero exit
Returns:
Exit code
"""
@classmethod
def initialize(cls, io_loop=None):
"""Initialize subprocess support."""
@classmethod
def uninitialize(cls):
"""Clean up subprocess support."""Locale and Internationalization
Internationalization support with CSV and gettext backends for translations and locale-aware formatting.
def get(code: str):
"""
Get locale by code.
Args:
code: Locale code
Returns:
Locale object
"""
def set_default_locale(code: str):
"""
Set default locale.
Args:
code: Locale code
"""
def load_translations(directory: str, encoding: str = None):
"""
Load translations from CSV files.
Args:
directory: Translations directory
encoding: File encoding
"""
def load_gettext_translations(directory: str, domain: str):
"""
Load gettext translations.
Args:
directory: Translations directory
domain: Translation domain
"""
def get_supported_locales(locale_path: str) -> List[str]:
"""
Get list of supported locales.
Args:
locale_path: Path to locale files
Returns:
List of locale codes
"""
class Locale:
"""Locale object for translations and formatting."""
def __init__(self, code: str, translations: Dict[str, str]):
"""
Initialize locale.
Args:
code: Locale code
translations: Translation dictionary
"""
@classmethod
def get_closest(cls, *locale_codes):
"""Get closest matching locale."""
@classmethod
def get(cls, code: str):
"""Get locale by code."""
def translate(self, message: str, plural_message: str = None, count: int = None) -> str:
"""
Translate message.
Args:
message: Message to translate
plural_message: Plural form
count: Count for plural selection
Returns:
Translated message
"""
def pgettext(self, context: str, message: str, plural_message: str = None, count: int = None) -> str:
"""
Translate message with context.
Args:
context: Translation context
message: Message to translate
plural_message: Plural form
count: Count for plural selection
Returns:
Translated message
"""
def format_date(self, date, gmt_offset: int = 0, relative: bool = True, shorter: bool = False, full_format: bool = False) -> str:
"""Format date for locale."""
def format_day(self, date, gmt_offset: int = 0, dow: bool = True) -> str:
"""Format day for locale."""
def list(self, parts: List[str]) -> str:
"""Format list for locale."""
def friendly_number(self, value: int) -> str:
"""Format number in friendly way."""
@property
def code(self) -> str:
"""Get locale code."""
@property
def name(self) -> str:
"""Get locale name."""General Utilities
Miscellaneous utility functions and classes for object handling, imports, and data structures.
def import_object(name: str):
"""
Import object by name.
Args:
name: Dotted object name
Returns:
Imported object
"""
def exec_in(code, glob, loc=None):
"""
Execute code in namespace.
Args:
code: Code to execute
glob: Global namespace
loc: Local namespace
"""
def raise_exc_info(exc_info):
"""
Raise exception from exc_info tuple.
Args:
exc_info: Exception info tuple
"""
def errno_from_exception(e) -> int:
"""
Get errno from exception.
Args:
e: Exception object
Returns:
Error number or None
"""
def re_unescape(s: str) -> str:
"""
Unescape regex string.
Args:
s: Escaped regex string
Returns:
Unescaped string
"""
def timedelta_to_seconds(td) -> float:
"""
Convert timedelta to seconds.
Args:
td: timedelta object
Returns:
Seconds as float
"""
class ObjectDict(dict):
"""Dictionary with attribute access."""
def __getattr__(self, name):
try:
return self[name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
self[name] = value
class GzipDecompressor:
"""Streaming gzip decompression."""
def __init__(self):
"""Initialize decompressor."""
def decompress(self, data: bytes, max_length: int = 0) -> bytes:
"""
Decompress data chunk.
Args:
data: Compressed data
max_length: Maximum output length
Returns:
Decompressed data
"""
@property
def unconsumed_tail(self) -> bytes:
"""Get unconsumed input data."""
@property
def eof(self) -> bool:
"""Check if end of stream reached."""
class Configurable:
"""Base class for configurable objects."""
@classmethod
def configurable_base(cls):
"""Get configurable base class."""
@classmethod
def configurable_default(cls):
"""Get default implementation class."""
@classmethod
def configure(cls, impl, **kwargs):
"""
Configure implementation class.
Args:
impl: Implementation class name or class
**kwargs: Configuration arguments
"""
@classmethod
def configured_class(cls):
"""Get currently configured class."""
class ArgReplacer:
"""Replace function arguments."""
def __init__(self, func, name: str):
"""
Initialize arg replacer.
Args:
func: Function to modify
name: Argument name to replace
"""
def get_old_value(self, args, kwargs):
"""Get old argument value."""
def replace(self, new_value, args, kwargs):
"""Replace argument with new value."""Usage Examples
Command-line Options
import tornado.options
# Define options
tornado.options.define("port", default=8888, help="run on the given port", type=int)
tornado.options.define("debug", default=False, help="run in debug mode", type=bool)
tornado.options.define("config", help="config file path")
# Parse command line
tornado.options.parse_command_line()
# Use options
print(f"Starting server on port {tornado.options.options.port}")
if tornado.options.options.debug:
print("Debug mode enabled")
# Parse config file if specified
if tornado.options.options.config:
tornado.options.parse_config_file(tornado.options.options.config)String Processing
import tornado.escape
# HTML escaping
html = tornado.escape.xhtml_escape('<script>alert("xss")</script>')
print(html) # <script>alert("xss")</script>
# URL encoding
url = tornado.escape.url_escape("hello world")
print(url) # hello%20world
# JSON handling
data = {"name": "John", "age": 30}
json_str = tornado.escape.json_encode(data)
parsed = tornado.escape.json_decode(json_str)
# Linkify URLs
text = "Visit https://www.example.com for more info"
linked = tornado.escape.linkify(text)
print(linked) # Visit <a href="https://www.example.com">https://www.example.com</a> for more infoProcess Management
import tornado.process
import tornado.web
import tornado.ioloop
def main():
# Fork multiple processes
tornado.process.fork_processes(0) # 0 = number of CPUs
app = tornado.web.Application([
(r"/", MainHandler),
])
app.listen(8888)
tornado.ioloop.IOLoop.current().start()
if __name__ == "__main__":
main()Internationalization
import tornado.locale
# Load translations
tornado.locale.load_translations("locale")
# Get locale
locale = tornado.locale.get("en_US")
# Translate messages
message = locale.translate("Hello, world!")
plural = locale.translate("You have %(count)d messages",
"You have %(count)d messages",
count=5)
# Format dates
import datetime
date_str = locale.format_date(datetime.datetime.now())Utility Objects
import tornado.util
# Object dictionary
obj = tornado.util.ObjectDict()
obj.name = "John"
obj.age = 30
print(obj.name) # Attribute access
print(obj['age']) # Dictionary access
# Import by name
cls = tornado.util.import_object("tornado.web.RequestHandler")
# Gzip decompression
decompressor = tornado.util.GzipDecompressor()
chunk1 = decompressor.decompress(compressed_data)Types
# Option types
OptionType = Union[type, Callable[[str], Any]]
OptionValue = Any
OptionDict = Dict[str, OptionValue]
# Locale types
LocaleCode = str
TranslationDict = Dict[str, str]
# Process types
ProcessId = int
ExitCode = int
# Utility types
ImportPath = str
ExcInfo = Tuple[type, Exception, Any]Exceptions
class Error(Exception):
"""Base exception for options module."""
class ConfigError(Error):
"""Exception for configuration errors."""
class LocaleNotFoundError(Exception):
"""Exception when locale is not found."""
def __init__(self, code: str):
"""
Initialize locale not found error.
Args:
code: Locale code that was not found
"""
class ProcessExitError(Exception):
"""Exception for subprocess exit errors."""
def __init__(self, exit_code: int):
"""
Initialize process exit error.
Args:
exit_code: Process exit code
"""Install with Tessl CLI
npx tessl i tessl/pypi-tornado