tessl/pypi-twisted
An asynchronous networking framework written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
logging.mddocs/
Logging System
Modern structured logging system with event-based architecture, filtering, and multiple output formats for comprehensive application logging. Twisted provides both legacy and modern logging systems.
Capabilities
Modern Structured Logging
Event-based logging system with structured data and flexible formatting.
class Logger:
"""
Main logging class for structured event logging.
Attributes:
- namespace: Logger namespace (str)
- source: Logger source object
"""
namespace = None
source = None
def __init__(self, namespace=None, source=None, observer=None):
"""
Args:
namespace (str): Logger namespace
source: Source object for logger
observer: Log observer (defaults to global)
"""
def debug(self, format=None, **kwargs):
"""
Log debug message.
Args:
format (str): Message format string
**kwargs: Format parameters and metadata
"""
def info(self, format=None, **kwargs):
"""
Log info message.
Args:
format (str): Message format string
**kwargs: Format parameters and metadata
"""
def warn(self, format=None, **kwargs):
"""
Log warning message.
Args:
format (str): Message format string
**kwargs: Format parameters and metadata
"""
def error(self, format=None, **kwargs):
"""
Log error message.
Args:
format (str): Message format string
**kwargs: Format parameters and metadata
"""
def critical(self, format=None, **kwargs):
"""
Log critical message.
Args:
format (str): Message format string
**kwargs: Format parameters and metadata
"""
def failure(self, format=None, failure=None, **kwargs):
"""
Log failure with traceback.
Args:
format (str): Message format string
failure: Failure object (defaults to current exception)
**kwargs: Format parameters and metadata
"""
class LogLevel:
"""
Log level constants and validation.
Constants:
- debug: Debug level
- info: Info level
- warn: Warning level
- error: Error level
- critical: Critical level
"""
debug = LogLevel._levelWithName('debug')
info = LogLevel._levelWithName('info')
warn = LogLevel._levelWithName('warn')
error = LogLevel._levelWithName('error')
critical = LogLevel._levelWithName('critical')
@classmethod
def levelWithName(cls, name):
"""
Get log level by name.
Args:
name (str): Level name
Returns:
LogLevel: Log level object
"""
class LogPublisher:
"""
Event publishing system for log observers.
Attributes:
- observers: List of registered observers
"""
observers = None
def __init__(self, *observers):
"""
Args:
*observers: Initial log observers
"""
def addObserver(self, observer):
"""
Add log observer.
Args:
observer: Log observer callable
"""
def removeObserver(self, observer):
"""
Remove log observer.
Args:
observer: Log observer to remove
"""
def msg(self, *message, **kwargs):
"""
Publish log message.
Args:
*message: Message components
**kwargs: Event metadata
"""
# Global logger instance
log = Logger()Modern Logging Usage Example:
from twisted.logger import Logger
# Create logger with namespace
log = Logger("myapp.module")
# Basic logging
log.info("Server starting on port {port}", port=8080)
log.warn("High memory usage: {usage}%", usage=85)
log.error("Database connection failed", db_host="localhost")
# Structured logging with metadata
log.info("User login",
user_id=12345,
ip_address="192.168.1.100",
success=True,
duration=0.5)
# Failure logging
try:
risky_operation()
except Exception:
log.failure("Operation failed") # Includes traceback
# Context-specific logging
user_log = Logger("myapp.user", source={"user_id": 12345})
user_log.info("Profile updated", fields=["email", "name"])Log Observers
Output destinations and formatting for log events.
class FileLogObserver:
"""
Log observer that writes to files.
"""
def __init__(self, outFile, formatEvent=None):
"""
Args:
outFile: File-like object to write to
formatEvent: Event formatting function
"""
def __call__(self, event):
"""
Process log event.
Args:
event (dict): Log event data
"""
class FilteringLogObserver:
"""
Log observer that filters events before forwarding.
"""
def __init__(self, observer, filters=None, defaultLogLevel=None):
"""
Args:
observer: Target observer
filters: List of filter predicates
defaultLogLevel: Default level for events without level
"""
def __call__(self, event):
"""
Filter and forward log event.
Args:
event (dict): Log event data
"""
class LimitedHistoryLogObserver:
"""
Log observer that keeps recent history in memory.
"""
def __init__(self, size=100):
"""
Args:
size (int): Maximum number of events to keep
"""
def __call__(self, event):
"""
Store log event in history.
Args:
event (dict): Log event data
"""
def getHistory(self):
"""
Get recent log history.
Returns:
list: Recent log events
"""
class LoggingFile:
"""
File-like object that writes to logger.
"""
def __init__(self, logger, level=LogLevel.info):
"""
Args:
logger: Logger to write to
level: Log level for messages
"""
def write(self, data):
"""
Write data to logger.
Args:
data (str): Data to log
"""
def flush(self):
"""Flush any buffered data."""Event Formatting
Functions for formatting log events into human-readable output.
def formatEvent(event):
"""
Format log event as string.
Args:
event (dict): Log event data
Returns:
str: Formatted event string
"""
def formatEventAsClassicLogText(event):
"""
Format event in classic log text format.
Args:
event (dict): Log event data
Returns:
str: Classic format log line
"""
def formatTime(when, timeFormat=None, default="-"):
"""
Format timestamp for logging.
Args:
when: Timestamp (float seconds since epoch)
timeFormat (str): Time format string
default (str): Default if formatting fails
Returns:
str: Formatted timestamp
"""
def eventAsText(event, includeTraceback=True):
"""
Convert log event to text representation.
Args:
event (dict): Log event data
includeTraceback (bool): Include traceback if present
Returns:
str: Text representation
"""
def eventFromJSON(jsonText):
"""
Parse log event from JSON text.
Args:
jsonText (str): JSON event data
Returns:
dict: Log event
"""
def eventAsJSON(event):
"""
Serialize log event as JSON.
Args:
event (dict): Log event data
Returns:
str: JSON representation
"""Observer Usage Example:
from twisted.logger import Logger, FileLogObserver, FilteringLogObserver
import sys
# File observer
log_file = open("/var/log/myapp.log", "a")
file_observer = FileLogObserver(log_file)
# Filtered observer (only warnings and errors)
def warning_filter(event):
level = event.get('log_level')
return level and level >= LogLevel.warn
filtered_observer = FilteringLogObserver(
observer=file_observer,
filters=[warning_filter]
)
# Console observer
console_observer = FileLogObserver(sys.stdout)
# Set up global logging
from twisted.logger import globalLogPublisher
globalLogPublisher.addObserver(console_observer)
globalLogPublisher.addObserver(filtered_observer)
# Use logger
log = Logger("myapp")
log.info("This goes to console only")
log.error("This goes to both console and file")Context Management
Context managers for logging operations and scoped logging.
class Operation:
"""
Context manager for logging operations with start/end events.
"""
def __init__(self, logger, message, level=LogLevel.info, **fields):
"""
Args:
logger: Logger to use
message (str): Operation description
level: Log level
**fields: Additional context fields
"""
def __enter__(self):
"""
Start operation logging.
Returns:
Operation: Self for chaining
"""
def __exit__(self, exc_type, exc_val, exc_tb):
"""
End operation logging.
Args:
exc_type: Exception type (if any)
exc_val: Exception value
exc_tb: Exception traceback
"""
def succeed(self, **fields):
"""
Mark operation as successful.
Args:
**fields: Success context fields
"""
def fail(self, **fields):
"""
Mark operation as failed.
Args:
**fields: Failure context fields
"""Context Usage Example:
from twisted.logger import Logger, Operation
log = Logger("myapp.database")
# Operation context manager
with Operation(log, "Database migration", version="1.2.3") as op:
try:
# Perform migration
migrate_schema()
op.succeed(tables_modified=5)
except Exception as e:
op.fail(error=str(e))
raise
# Manual operation control
op = Operation(log, "User authentication", user_id=12345)
op.__enter__()
try:
if authenticate_user():
op.succeed(method="password")
else:
op.fail(reason="invalid_password")
finally:
op.__exit__(None, None, None)Legacy Logging (twisted.python.log)
Backward-compatible logging system for older Twisted code.
# Legacy logging functions from twisted.python.log
def msg(*message, **kwargs):
"""
Log a message using legacy system.
Args:
*message: Message components
**kwargs: Additional data
"""
def err(failure=None, _why=None):
"""
Log an error using legacy system.
Args:
failure: Failure object (defaults to current exception)
_why (str): Error description
"""
def startLogging(file, setStdout=1):
"""
Start logging to a file.
Args:
file: File-like object or filename
setStdout (bool): Redirect stdout to log
"""
def startLoggingWithObserver(observer, setStdout=1):
"""
Start logging with custom observer.
Args:
observer: Log observer function
setStdout (bool): Redirect stdout to log
"""
def addObserver(observer):
"""
Add legacy log observer.
Args:
observer: Observer function
"""
def removeObserver(observer):
"""
Remove legacy log observer.
Args:
observer: Observer function to remove
"""
class FileLogObserver:
"""
Legacy file log observer.
"""
def __init__(self, f):
"""
Args:
f: File-like object
"""
def __call__(self, eventDict):
"""
Process legacy log event.
Args:
eventDict (dict): Legacy event data
"""
class PythonLoggingObserver:
"""
Observer that forwards to Python's logging module.
"""
def __init__(self, loggerName='twisted'):
"""
Args:
loggerName (str): Python logger name
"""Legacy Logging Example:
from twisted.python import log
import sys
# Start legacy logging
log.startLogging(sys.stdout)
# Log messages
log.msg("Server starting", port=8080)
log.msg("Connection established", client="192.168.1.100")
# Log errors
try:
risky_operation()
except:
log.err() # Logs current exception
# Custom observer
def my_observer(event):
print(f"Custom: {event}")
log.addObserver(my_observer)
log.msg("This will go to custom observer too")Integration with Twisted Components
Logging integration patterns for Twisted protocols and services.
# Protocol logging
from twisted.internet import protocol
from twisted.logger import Logger
class MyProtocol(protocol.Protocol):
def __init__(self):
self.log = Logger(self.__class__.__name__, source=self)
def connectionMade(self):
self.log.info("Connection established",
peer=str(self.transport.getPeer()))
def dataReceived(self, data):
self.log.debug("Data received",
bytes=len(data),
data=data[:100]) # Log first 100 bytes
# Service logging
from twisted.application import service
class MyService(service.Service):
def __init__(self):
self.log = Logger(self.__class__.__name__)
def startService(self):
service.Service.startService(self)
self.log.info("Service started")
def stopService(self):
self.log.info("Service stopping")
return service.Service.stopService(self)Log Analysis and Monitoring
Utilities for analyzing and monitoring log output.
# Event filtering
def jsonFileLogObserver(outFile):
"""
Create observer that writes JSON events to file.
Args:
outFile: Output file
Returns:
Observer function
"""
def eventsFromJSONLogFile(inFile):
"""
Read log events from JSON log file.
Args:
inFile: Input file
Yields:
dict: Log events
"""
# Custom filters
def predicateLogObserver(observer, predicate):
"""
Create filtering observer with custom predicate.
Args:
observer: Target observer
predicate: Filter predicate function
Returns:
Filtering observer
"""
def levelLogObserver(observer, level):
"""
Create observer that filters by log level.
Args:
observer: Target observer
level: Minimum log level
Returns:
Filtering observer
"""Install with Tessl CLI
npx tessl i tessl/pypi-twisted