Diagnostics Module¶

The rite.diagnostics module provides logging, profiling, error handling, and debugging utilities.

Overview¶

diagnostics ¶

Diagnostics Module¶

Comprehensive diagnostics and monitoring utilities.

This module provides utilities for logging, error handling, profiling, debugging, and metrics collection.

Submodules¶

logging: Structured logging, file/console output, contextual logging
errors: Exception handling, retry logic, traceback formatting
profiling: Timing, memory profiling, call counting
debugging: Object inspection, function tracing, state dumping
metrics: Counters, gauges, histograms, timers

Examples¶

Logging: >>> from rite.diagnostics import logging_structured >>> logging_structured("INFO", "Application started", user="admin")

Error Handling

from rite.diagnostics import errors_retry @errors_retry(max_attempts=3) ... def api_call(): ... pass

Profiling

from rite.diagnostics import profiling_timer @profiling_timer() ... def slow_function(): ... pass

Debugging

from rite.diagnostics import debugging_trace @debugging_trace() ... def calculate(x, y): ... return x + y

Metrics

from rite.diagnostics import metrics_counter counter = metrics_counter("requests") counter.increment()

Classes¶

ErrorHandler ¶

ErrorHandler(log_file: str | None = 'error_log.txt')

Error Handler Class¶

A class to handle and log errors in the system. It provides methods for logging errors, raising alerts, and retrying operations.

Attributes¶

log_file : str | None The path to the log file for error logging.

Initializes the ErrorHandler with a specified log file.

Parameters:¶

log_file : str | None The path to the log file (default: 'error_log.txt').

Functions¶

log_error ¶

log_error(error: str) -> None

Logs an error message to the log file.

Parameters:¶

error : str The error message to log.

raise_alert ¶

raise_alert(error: str) -> None

Raises an alert for critical errors and logs the alert message.

Parameters:¶

error : str The error message to alert on.

retry_operation ¶

retry_operation(operation: Callable, retries: int = 3, backoff: float = 1.0) -> bool

Retries a specified operation a given number of times with optional exponential backoff.

Parameters:¶

operation : Callable The operation (function) to retry. retries : int, optional The number of retry attempts (default: 3). backoff : float, optional The initial backoff time in seconds between retries (default: 1.0).

Returns¶

bool: True if the operation succeeded, False if all retries failed.

Logger ¶

Logger(name: str, log_file: str | None = None)

Logger Class¶

A configurable logger that supports logging to both console and file outputs with various log levels. This class wraps Python's built-in logging module, providing an easy interface to integrate logging into applications.

Attributes¶

logger : logging.Logger The underlying logger instance from Python's logging module. log_file : str | None Path to the log file if file logging is enabled.

Initializes a new Logger instance with the specified name and optional log file.

Parameters:¶

name : str The name of the logger, typically the module or application name. log_file : str | None Path to a log file for file logging. If None, logging to file is disabled.

Example:¶

logger = Logger(name="MyApp", log_file="app.log") logger.info("Application started")

Functions¶

clear_log ¶

clear_log() -> None

Clears the log file content, if file logging is enabled.

critical ¶

critical(message: str) -> None

Logs a critical message.

Parameters:¶

message : str The critical message to log.

Example:¶

logger.critical("This is a critical message")

debug ¶

debug(message: str) -> None

Logs a debug message.

Parameters:¶

message : str The debug message to log.

Example:¶

logger.debug("This is a debug message")

error ¶

error(message: str) -> None

Logs an error message.

Parameters:¶

message : str The error message to log.

Example:¶

logger.error("This is an error message")

info ¶

info(message: str) -> None

Logs an informational message.

Parameters:¶

message : str The informational message to log.

Example:¶

logger.info("This is an info message")

log ¶

log(level: str, message: str) -> None

Logs a message with the specified level.

Parameters:¶

level : str The log level (e.g., "debug", "info", "warning", "error", "critical"). message : str The message to log.

warning ¶

warning(message: str) -> None

Logs a warning message.

Parameters:¶

message : str The warning message to log.

Example:¶

logger.warning("This is a warning message")

Modules¶

debugging ¶

Debugging Module¶

Debugging and inspection utilities.

This submodule provides utilities for inspecting objects, tracing function calls, and dumping variable states.

Examples¶

from rite.diagnostics.debugging import ( ... debugging_inspect, ... debugging_trace, ... debugging_dump ... ) @debugging_trace() ... def example(): ... pass

Modules¶

debugging_dump ¶

State Dumper¶

Dump variable states for debugging.

Examples¶

from rite.diagnostics.debugging import debugging_dump debugging_dump(x=1, y=2, name="test")

Functions¶

debugging_dump ¶

debugging_dump(*args: Any, **kwargs: Any) -> None

Print variables in formatted way for debugging.

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Positional arguments to dump.	`()`
`**kwargs`	`Any`	Named arguments to dump.	`{}`

Examples:

>>> debugging_dump(42, "hello", status="active")
ARG 0: 42
ARG 1: hello
status: active
>>> data = {"key": "value"}
>>> debugging_dump(data=data)
data: {'key': 'value'}

debugging_inspect ¶

Variable Inspector¶

Inspect variables and their attributes.

Examples¶

from rite.diagnostics.debugging import debugging_inspect obj = {"key": "value"} debugging_inspect(obj)

Functions¶

debugging_inspect ¶

debugging_inspect(obj: Any, show_private: bool = False) -> dict[str, Any]

Inspect object attributes and methods.

Parameters:

Name	Type	Description	Default
`obj`	`Any`	Object to inspect.	required
`show_private`	`bool`	Include private attributes (starting with _).	`False`

Returns:

Type	Description
`dict[str, Any]`	Dictionary with object information.

Examples:

>>> class Example:
...     def __init__(self):
...         self.value = 42
>>> info = debugging_inspect(Example())
>>> "value" in info["attributes"]
True

debugging_locals ¶

Locals Dumper¶

Dump all local variables in current scope.

Examples¶

from rite.diagnostics.debugging import debugging_locals def example(): ... x = 1 ... y = 2 ... debugging_locals()

Functions¶

debugging_locals ¶

debugging_locals(show_private: bool = False) -> dict[str, Any]

Dump local variables from calling scope.

Parameters:

Name	Type	Description	Default
`show_private`	`bool`	Include private variables (starting with _).	`False`

Returns:

Type	Description
`dict[str, Any]`	Dictionary of local variables.

Examples:

>>> def test():
...     x = 42
...     y = "hello"
...     locals_dict = debugging_locals()
...     return "x" in locals_dict
>>> test()
True

Notes

This function inspects the caller's frame to get locals.

debugging_trace ¶

Function Tracer¶

Trace function calls with arguments and returns.

Examples¶

from rite.diagnostics.debugging import debugging_trace @debugging_trace() ... def add(a, b): ... return a + b

Functions¶

debugging_trace ¶

debugging_trace(show_args: bool = True, show_return: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to trace function calls.

Parameters:

Name	Type	Description	Default
`show_args`	`bool`	Show function arguments.	`True`
`show_return`	`bool`	Show return value.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with tracing.

Examples:

>>> @debugging_trace()
... def multiply(x, y):
...     return x * y
>>> @debugging_trace(show_args=False)
... def process():
...     pass

debugging_dump ¶

State Dumper¶

Dump variable states for debugging.

Examples¶

from rite.diagnostics.debugging import debugging_dump debugging_dump(x=1, y=2, name="test")

Functions¶

debugging_dump ¶

debugging_dump(*args: Any, **kwargs: Any) -> None

Print variables in formatted way for debugging.

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Positional arguments to dump.	`()`
`**kwargs`	`Any`	Named arguments to dump.	`{}`

Examples:

>>> debugging_dump(42, "hello", status="active")
ARG 0: 42
ARG 1: hello
status: active
>>> data = {"key": "value"}
>>> debugging_dump(data=data)
data: {'key': 'value'}

debugging_inspect ¶

Variable Inspector¶

Inspect variables and their attributes.

Examples¶

from rite.diagnostics.debugging import debugging_inspect obj = {"key": "value"} debugging_inspect(obj)

Functions¶

debugging_inspect ¶

debugging_inspect(obj: Any, show_private: bool = False) -> dict[str, Any]

Inspect object attributes and methods.

Parameters:

Name	Type	Description	Default
`obj`	`Any`	Object to inspect.	required
`show_private`	`bool`	Include private attributes (starting with _).	`False`

Returns:

Type	Description
`dict[str, Any]`	Dictionary with object information.

Examples:

>>> class Example:
...     def __init__(self):
...         self.value = 42
>>> info = debugging_inspect(Example())
>>> "value" in info["attributes"]
True

debugging_locals ¶

Locals Dumper¶

Dump all local variables in current scope.

Examples¶

from rite.diagnostics.debugging import debugging_locals def example(): ... x = 1 ... y = 2 ... debugging_locals()

Functions¶

debugging_locals ¶

debugging_locals(show_private: bool = False) -> dict[str, Any]

Dump local variables from calling scope.

Parameters:

Name	Type	Description	Default
`show_private`	`bool`	Include private variables (starting with _).	`False`

Returns:

Type	Description
`dict[str, Any]`	Dictionary of local variables.

Examples:

>>> def test():
...     x = 42
...     y = "hello"
...     locals_dict = debugging_locals()
...     return "x" in locals_dict
>>> test()
True

Notes

This function inspects the caller's frame to get locals.

debugging_trace ¶

Function Tracer¶

Trace function calls with arguments and returns.

Examples¶

from rite.diagnostics.debugging import debugging_trace @debugging_trace() ... def add(a, b): ... return a + b

Functions¶

debugging_trace ¶

debugging_trace(show_args: bool = True, show_return: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to trace function calls.

Parameters:

Name	Type	Description	Default
`show_args`	`bool`	Show function arguments.	`True`
`show_return`	`bool`	Show return value.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with tracing.

Examples:

>>> @debugging_trace()
... def multiply(x, y):
...     return x * y
>>> @debugging_trace(show_args=False)
... def process():
...     pass

error_handler ¶

Error Handling Module¶

This module provides the ErrorHandler class for managing and logging errors in the system. It supports error logging, raising alerts, and retrying failed operations with customizable retry logic.

Classes:¶

ErrorHandler: Handles error logging, alerting, and retrying operations.

Features:¶

Logs errors to a specified log file.
Raises alerts for critical errors.
Retries failed operations with configurable retry limits and exponential backoff.

Classes¶

ErrorHandler ¶

ErrorHandler(log_file: str | None = 'error_log.txt')

Error Handler Class¶

A class to handle and log errors in the system. It provides methods for logging errors, raising alerts, and retrying operations.

Attributes¶

log_file : str | None The path to the log file for error logging.

Initializes the ErrorHandler with a specified log file.

Parameters:¶

log_file : str | None The path to the log file (default: 'error_log.txt').

Functions¶

log_error ¶

log_error(error: str) -> None

Logs an error message to the log file.

Parameters:¶

error : str The error message to log.

raise_alert ¶

raise_alert(error: str) -> None

Raises an alert for critical errors and logs the alert message.

Parameters:¶

error : str The error message to alert on.

retry_operation ¶

retry_operation(operation: Callable, retries: int = 3, backoff: float = 1.0) -> bool

Retries a specified operation a given number of times with optional exponential backoff.

Parameters:¶

operation : Callable The operation (function) to retry. retries : int, optional The number of retry attempts (default: 3). backoff : float, optional The initial backoff time in seconds between retries (default: 1.0).

Returns¶

bool: True if the operation succeeded, False if all retries failed.

Functions¶

failing_operation ¶

failing_operation()

A simulated operation that raises a RuntimeError.

errors ¶

Error Handling Module¶

Error handling utilities with retry, catching, and formatting.

This submodule provides utilities for handling errors gracefully: retry decorators, exception context managers, traceback formatting, and exception chain analysis.

Examples¶

from rite.diagnostics.errors import errors_retry, errors_catch @errors_retry(max_attempts=3) ... def flaky_function(): ... pass with errors_catch(): ... risky_operation()

Modules¶

errors_catch ¶

Exception Context Manager¶

Context manager to catch and handle exceptions gracefully.

Examples¶

from rite.diagnostics.errors import errors_catch with errors_catch(): ... risky_operation()

Functions¶

errors_catch ¶

errors_catch(exceptions: tuple[type[Exception], ...] = (Exception,), handler: Callable[[Exception], Any] | None = None, reraise: bool = False, logger: Logger | None = None) -> Generator[None, None, None]

Context manager to catch and handle exceptions.

Parameters:

Name	Type	Description	Default
`exceptions`	`tuple[type[Exception], ...]`	Tuple of exceptions to catch.	`(Exception,)`
`handler`	`Callable[[Exception], Any] \| None`	Optional handler function called with exception.	`None`
`reraise`	`bool`	If True, reraise exception after handling.	`False`
`logger`	`Logger \| None`	Optional logger to log exceptions.	`None`

Yields:

Type	Description
`None`	None.

Examples:

>>> with errors_catch():
...     1 / 0
>>> with errors_catch(handler=lambda e: print(f"Error: {e}")):
...     risky_operation()
>>> with errors_catch((ValueError, KeyError), reraise=True):
...     parse_data()

errors_format_traceback ¶

Traceback Formatter¶

Format exception tracebacks for better readability.

Examples¶

from rite.diagnostics.errors import errors_format_traceback try: ... 1 / 0 ... except Exception as e: ... print(errors_format_traceback(e))

Functions¶

errors_format_traceback ¶

errors_format_traceback(exception: BaseException, include_locals: bool = False) -> str

Format exception traceback as string.

Parameters:

Name	Type	Description	Default
`exception`	`BaseException`	Exception to format.	required
`include_locals`	`bool`	If True, include local variables.	`False`

Returns:

Type	Description
`str`	Formatted traceback string.

Examples:

>>> try:
...     raise ValueError("test error")
... except ValueError as e:
...     tb = errors_format_traceback(e)
...     "ValueError" in tb
True

errors_get_chain ¶

Exception Chain¶

Get exception chain (cause hierarchy).

Examples¶

from rite.diagnostics.errors import errors_get_chain try: ... try: ... 1 / 0 ... except ZeroDivisionError as e: ... raise ValueError("Wrapped") from e ... except ValueError as e: ... chain = errors_get_chain(e)

Functions¶

errors_get_chain ¶

errors_get_chain(exception: BaseException) -> list[BaseException]

Get exception chain (cause hierarchy).

Parameters:

Name	Type	Description	Default
`exception`	`BaseException`	Exception to get chain for.	required

Returns:

Type	Description
`list[BaseException]`	List of exceptions from root cause to final exception.

Examples:

>>> try:
...     try:
...         raise ValueError("root")
...     except ValueError as e:
...         raise KeyError("wrapped") from e
... except KeyError as e:
...     chain = errors_get_chain(e)
...     len(chain)
2

errors_retry ¶

Retry Decorator¶

Decorator to retry function calls on failure.

Examples¶

from rite.diagnostics.errors import errors_retry @errors_retry(max_attempts=3, delay=1.0) ... def unstable_function(): ... pass

Functions¶

errors_retry ¶

errors_retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0, exceptions: tuple[type[Exception], ...] = (Exception,)) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to retry function on failure.

Parameters:

Name	Type	Description	Default
`max_attempts`	`int`	Maximum retry attempts.	`3`
`delay`	`float`	Initial delay between retries (seconds).	`1.0`
`backoff`	`float`	Multiplier for delay on each retry.	`2.0`
`exceptions`	`tuple[type[Exception], ...]`	Tuple of exceptions to catch.	`(Exception,)`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with retry logic.

Examples:

>>> @errors_retry(max_attempts=3)
... def flaky_api_call():
...     pass
>>> @errors_retry(max_attempts=5, delay=0.5, backoff=1.5)
... def database_query():
...     pass

errors_catch ¶

Exception Context Manager¶

Context manager to catch and handle exceptions gracefully.

Examples¶

from rite.diagnostics.errors import errors_catch with errors_catch(): ... risky_operation()

Functions¶

errors_catch ¶

errors_catch(exceptions: tuple[type[Exception], ...] = (Exception,), handler: Callable[[Exception], Any] | None = None, reraise: bool = False, logger: Logger | None = None) -> Generator[None, None, None]

Context manager to catch and handle exceptions.

Parameters:

Name	Type	Description	Default
`exceptions`	`tuple[type[Exception], ...]`	Tuple of exceptions to catch.	`(Exception,)`
`handler`	`Callable[[Exception], Any] \| None`	Optional handler function called with exception.	`None`
`reraise`	`bool`	If True, reraise exception after handling.	`False`
`logger`	`Logger \| None`	Optional logger to log exceptions.	`None`

Yields:

Type	Description
`None`	None.

Examples:

>>> with errors_catch():
...     1 / 0
>>> with errors_catch(handler=lambda e: print(f"Error: {e}")):
...     risky_operation()
>>> with errors_catch((ValueError, KeyError), reraise=True):
...     parse_data()

errors_format_traceback ¶

Traceback Formatter¶

Format exception tracebacks for better readability.

Examples¶

from rite.diagnostics.errors import errors_format_traceback try: ... 1 / 0 ... except Exception as e: ... print(errors_format_traceback(e))

Functions¶

errors_format_traceback ¶

errors_format_traceback(exception: BaseException, include_locals: bool = False) -> str

Format exception traceback as string.

Parameters:

Name	Type	Description	Default
`exception`	`BaseException`	Exception to format.	required
`include_locals`	`bool`	If True, include local variables.	`False`

Returns:

Type	Description
`str`	Formatted traceback string.

Examples:

>>> try:
...     raise ValueError("test error")
... except ValueError as e:
...     tb = errors_format_traceback(e)
...     "ValueError" in tb
True

errors_get_chain ¶

Exception Chain¶

Get exception chain (cause hierarchy).

Examples¶

from rite.diagnostics.errors import errors_get_chain try: ... try: ... 1 / 0 ... except ZeroDivisionError as e: ... raise ValueError("Wrapped") from e ... except ValueError as e: ... chain = errors_get_chain(e)

Functions¶

errors_get_chain ¶

errors_get_chain(exception: BaseException) -> list[BaseException]

Get exception chain (cause hierarchy).

Parameters:

Name	Type	Description	Default
`exception`	`BaseException`	Exception to get chain for.	required

Returns:

Type	Description
`list[BaseException]`	List of exceptions from root cause to final exception.

Examples:

>>> try:
...     try:
...         raise ValueError("root")
...     except ValueError as e:
...         raise KeyError("wrapped") from e
... except KeyError as e:
...     chain = errors_get_chain(e)
...     len(chain)
2

errors_retry ¶

Retry Decorator¶

Decorator to retry function calls on failure.

Examples¶

from rite.diagnostics.errors import errors_retry @errors_retry(max_attempts=3, delay=1.0) ... def unstable_function(): ... pass

Functions¶

errors_retry ¶

errors_retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0, exceptions: tuple[type[Exception], ...] = (Exception,)) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to retry function on failure.

Parameters:

Name	Type	Description	Default
`max_attempts`	`int`	Maximum retry attempts.	`3`
`delay`	`float`	Initial delay between retries (seconds).	`1.0`
`backoff`	`float`	Multiplier for delay on each retry.	`2.0`
`exceptions`	`tuple[type[Exception], ...]`	Tuple of exceptions to catch.	`(Exception,)`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with retry logic.

Examples:

>>> @errors_retry(max_attempts=3)
... def flaky_api_call():
...     pass
>>> @errors_retry(max_attempts=5, delay=0.5, backoff=1.5)
... def database_query():
...     pass

logger ¶

Logger Module¶

This module provides the Logger class, which serves as a simple and configurable logging utility for applications within the Sense ecosystem. It supports logging to both the console and a file, includes timestamped messages, and provides different logging levels such as debug, info, warning, error, and critical.

Classes:¶

Logger: A configurable logger for managing log messages in applications.

Features:¶

Log to console and/or file.
Timestamped messages for traceability.
Logging levels: debug, info, warning, error, critical.
Clear log file utility.

Classes¶

Logger ¶

Logger(name: str, log_file: str | None = None)

Logger Class¶

A configurable logger that supports logging to both console and file outputs with various log levels. This class wraps Python's built-in logging module, providing an easy interface to integrate logging into applications.

Attributes¶

logger : logging.Logger The underlying logger instance from Python's logging module. log_file : str | None Path to the log file if file logging is enabled.

Initializes a new Logger instance with the specified name and optional log file.

Parameters:¶

name : str The name of the logger, typically the module or application name. log_file : str | None Path to a log file for file logging. If None, logging to file is disabled.

Example:¶

logger = Logger(name="MyApp", log_file="app.log") logger.info("Application started")

Functions¶

clear_log ¶

clear_log() -> None

Clears the log file content, if file logging is enabled.

critical ¶

critical(message: str) -> None

Logs a critical message.

Parameters:¶

message : str The critical message to log.

Example:¶

logger.critical("This is a critical message")

debug ¶

debug(message: str) -> None

Logs a debug message.

Parameters:¶

message : str The debug message to log.

Example:¶

logger.debug("This is a debug message")

error ¶

error(message: str) -> None

Logs an error message.

Parameters:¶

message : str The error message to log.

Example:¶

logger.error("This is an error message")

info ¶

info(message: str) -> None

Logs an informational message.

Parameters:¶

message : str The informational message to log.

Example:¶

logger.info("This is an info message")

log ¶

log(level: str, message: str) -> None

Logs a message with the specified level.

Parameters:¶

level : str The log level (e.g., "debug", "info", "warning", "error", "critical"). message : str The message to log.

warning ¶

warning(message: str) -> None

Logs a warning message.

Parameters:¶

message : str The warning message to log.

Example:¶

logger.warning("This is a warning message")

logging ¶

Logging Module¶

Logging utilities with file, console, structured, and contextual output.

This submodule provides various logging configurations for different use cases: file logging with rotation, console logging with colors, structured JSON logging, and context-aware logging.

Examples¶

from rite.diagnostics.logging import ( ... logging_to_file, ... logging_to_console, ... logging_structured ... ) file_logger = logging_to_file("app", "app.log") console_logger = logging_to_console("debug", colorize=True) struct_logger = logging_structured("api")

Modules¶

logging_structured ¶

Structured Logger¶

Logger with structured output (JSON, key-value pairs).

Examples¶

from rite.diagnostics.logging import logging_structured logger = logging_structured("app") logger.info("User logged in", user_id=123, ip="192.168.1.1")

Functions¶

logging_structured ¶

logging_structured(name: str, level: int = logging.INFO, json_format: bool = True) -> logging.Logger

Create logger with structured output.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`json_format`	`bool`	If True, output as JSON; else key=value format.	`True`

Returns:

Type	Description
`Logger`	Configured logger with structured formatter.

Examples:

>>> logger = logging_structured("app")
>>> logger.info("Event occurred", user_id=123)
>>> logger = logging_structured("api", json_format=False)
>>> logger.warning("Slow query", duration_ms=500)

logging_to_console ¶

Console Logger¶

Create logger that writes to console/stdout.

Examples¶

from rite.diagnostics.logging import logging_to_console logger = logging_to_console("app") logger.info("Application started")

Functions¶

logging_to_console ¶

logging_to_console(name: str, level: int = logging.INFO, format_string: str | None = None, colorize: bool = False) -> logging.Logger

Create logger that writes to console.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`format_string`	`str \| None`	Custom format string.	`None`
`colorize`	`bool`	If True, colorize output by level.	`False`

Returns:

Type	Description
`Logger`	Configured console logger.

Examples:

>>> logger = logging_to_console("app")
>>> logger.info("Hello world")
>>> logger = logging_to_console("debug", level=logging.DEBUG)
>>> logger.debug("Debug message")

logging_to_file ¶

File Logger¶

Create logger that writes to a file with rotation support.

Examples¶

from rite.diagnostics.logging import logging_to_file logger = logging_to_file("app", "app.log") logger.info("Application started")

Functions¶

logging_to_file ¶

logging_to_file(name: str, filepath: str, level: int = logging.INFO, max_bytes: int = 10485760, backup_count: int = 5, format_string: str | None = None) -> logging.Logger

Create logger that writes to a file with rotation.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`filepath`	`str`	Path to log file.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`max_bytes`	`int`	Max file size before rotation (default 10MB).	`10485760`
`backup_count`	`int`	Number of backup files to keep (default 5).	`5`
`format_string`	`str \| None`	Custom format string.	`None`

Returns:

Type	Description
`Logger`	Configured file logger.

Examples:

>>> logger = logging_to_file("app", "logs/app.log")
>>> logger.info("Starting application")
>>> logger = logging_to_file(
...     "api",
...     "logs/api.log",
...     max_bytes=5242880,
...     backup_count=3
... )

logging_with_context ¶

Context Logger¶

Logger that includes contextual information in all messages.

Examples¶

from rite.diagnostics.logging import logging_with_context logger = logging_with_context("app", request_id="abc123") logger.info("Processing request")

Functions¶

logging_with_context ¶

logging_with_context(name: str, level: int = logging.INFO, **context: Any) -> logging.LoggerAdapter

Create logger that includes context in all messages.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`**context`	`Any`	Context key-value pairs to include.	`{}`

Returns:

Type	Description
`LoggerAdapter`	Logger adapter with context.

Examples:

>>> logger = logging_with_context(
...     "api",
...     request_id="req123",
...     user_id=456
... )
>>> logger.info("Request processed")
>>> logger = logging_with_context(
...     "worker",
...     worker_id="w1",
...     queue="default"
... )

logging_structured ¶

Structured Logger¶

Logger with structured output (JSON, key-value pairs).

Examples¶

from rite.diagnostics.logging import logging_structured logger = logging_structured("app") logger.info("User logged in", user_id=123, ip="192.168.1.1")

Functions¶

logging_structured ¶

logging_structured(name: str, level: int = logging.INFO, json_format: bool = True) -> logging.Logger

Create logger with structured output.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`json_format`	`bool`	If True, output as JSON; else key=value format.	`True`

Returns:

Type	Description
`Logger`	Configured logger with structured formatter.

Examples:

>>> logger = logging_structured("app")
>>> logger.info("Event occurred", user_id=123)
>>> logger = logging_structured("api", json_format=False)
>>> logger.warning("Slow query", duration_ms=500)

logging_to_console ¶

Console Logger¶

Create logger that writes to console/stdout.

Examples¶

from rite.diagnostics.logging import logging_to_console logger = logging_to_console("app") logger.info("Application started")

Functions¶

logging_to_console ¶

logging_to_console(name: str, level: int = logging.INFO, format_string: str | None = None, colorize: bool = False) -> logging.Logger

Create logger that writes to console.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`format_string`	`str \| None`	Custom format string.	`None`
`colorize`	`bool`	If True, colorize output by level.	`False`

Returns:

Type	Description
`Logger`	Configured console logger.

Examples:

>>> logger = logging_to_console("app")
>>> logger.info("Hello world")
>>> logger = logging_to_console("debug", level=logging.DEBUG)
>>> logger.debug("Debug message")

logging_to_file ¶

File Logger¶

Create logger that writes to a file with rotation support.

Examples¶

from rite.diagnostics.logging import logging_to_file logger = logging_to_file("app", "app.log") logger.info("Application started")

Functions¶

logging_to_file ¶

logging_to_file(name: str, filepath: str, level: int = logging.INFO, max_bytes: int = 10485760, backup_count: int = 5, format_string: str | None = None) -> logging.Logger

Create logger that writes to a file with rotation.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`filepath`	`str`	Path to log file.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`max_bytes`	`int`	Max file size before rotation (default 10MB).	`10485760`
`backup_count`	`int`	Number of backup files to keep (default 5).	`5`
`format_string`	`str \| None`	Custom format string.	`None`

Returns:

Type	Description
`Logger`	Configured file logger.

Examples:

>>> logger = logging_to_file("app", "logs/app.log")
>>> logger.info("Starting application")
>>> logger = logging_to_file(
...     "api",
...     "logs/api.log",
...     max_bytes=5242880,
...     backup_count=3
... )

logging_with_context ¶

Context Logger¶

Logger that includes contextual information in all messages.

Examples¶

from rite.diagnostics.logging import logging_with_context logger = logging_with_context("app", request_id="abc123") logger.info("Processing request")

Functions¶

logging_with_context ¶

logging_with_context(name: str, level: int = logging.INFO, **context: Any) -> logging.LoggerAdapter

Create logger that includes context in all messages.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`**context`	`Any`	Context key-value pairs to include.	`{}`

Returns:

Type	Description
`LoggerAdapter`	Logger adapter with context.

Examples:

>>> logger = logging_with_context(
...     "api",
...     request_id="req123",
...     user_id=456
... )
>>> logger.info("Request processed")
>>> logger = logging_with_context(
...     "worker",
...     worker_id="w1",
...     queue="default"
... )

metrics ¶

Metrics Module¶

Performance and monitoring metrics.

This submodule provides metric classes for counters, gauges, histograms, and timers.

Examples¶

from rite.diagnostics.metrics import ( ... metrics_counter, ... metrics_gauge, ... metrics_histogram, ... metrics_timer ... ) counter = metrics_counter("requests") counter.increment()

Modules¶

metrics_counter ¶

Counter Metric¶

Counter that can only increase.

Examples¶

from rite.diagnostics.metrics import metrics_counter counter = metrics_counter("requests") counter.increment() counter.value 1

Classes¶

metrics_counter ¶

metrics_counter(name: str)

Counter metric that only increases.

Attributes:

Name	Type	Description
`name`		Metric name.
`value`	`float`	Current counter value.

Examples:

>>> counter = metrics_counter("api_calls")
>>> counter.increment()
>>> counter.increment(5)
>>> counter.value
6

Initialize counter.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

value property ¶

value: float

Get current value.

Functions¶

increment ¶

increment(amount: float = 1.0) -> None

Increment counter.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to add (must be non-negative).	`1.0`

Raises:

Type	Description
`ValueError`	If amount is negative.

Examples:

>>> counter = metrics_counter("requests")
>>> counter.increment()
>>> counter.increment(10)
>>> counter.value
11.0

reset ¶

reset() -> None

Reset counter to zero.

metrics_gauge ¶

Gauge Metric¶

Gauge that can increase or decrease.

Examples¶

from rite.diagnostics.metrics import metrics_gauge gauge = metrics_gauge("temperature") gauge.set(25.0) gauge.value 25.0

Classes¶

metrics_gauge ¶

metrics_gauge(name: str, initial_value: float = 0.0)

Gauge metric that can go up or down.

Attributes:

Name	Type	Description
`name`		Metric name.
`value`	`float`	Current gauge value.

Examples:

>>> gauge = metrics_gauge("memory_usage")
>>> gauge.set(100)
>>> gauge.increment(50)
>>> gauge.decrement(25)
>>> gauge.value
125.0

Initialize gauge.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required
`initial_value`	`float`	Starting value.	`0.0`

Attributes¶

value property ¶

value: float

Get current value.

Functions¶

decrement ¶

decrement(amount: float = 1.0) -> None

Decrease gauge value.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to subtract.	`1.0`

increment ¶

increment(amount: float = 1.0) -> None

Increase gauge value.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to add.	`1.0`

reset ¶

reset() -> None

Reset gauge to zero.

set ¶

set(value: float) -> None

Set gauge to specific value.

Parameters:

Name	Type	Description	Default
`value`	`float`	New value.	required

Examples:

>>> gauge = metrics_gauge("cpu")
>>> gauge.set(75.5)
>>> gauge.value
75.5

metrics_histogram ¶

Histogram Metric¶

Histogram to track distribution of values.

Examples¶

from rite.diagnostics.metrics import metrics_histogram hist = metrics_histogram("response_time") hist.observe(0.5) hist.count 1

Classes¶

metrics_histogram ¶

metrics_histogram(name: str)

Histogram metric for value distributions.

Attributes:

Name	Type	Description
`name`		Metric name.
`count`	`int`	Number of observations.
`sum`	`float`	Sum of all values.

Examples:

>>> hist = metrics_histogram("latency")
>>> hist.observe(1.0)
>>> hist.observe(2.0)
>>> hist.observe(3.0)
>>> hist.mean
2.0

Initialize histogram.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

count property ¶

count: int

Get number of observations.

mean property ¶

mean: float

Get mean of values.

median property ¶

median: float

Get median of values.

sum property ¶

sum: float

Get sum of all values.

Functions¶

observe ¶

observe(value: float) -> None

Record observation.

Parameters:

Name	Type	Description	Default
`value`	`float`	Value to observe.	required

Examples:

>>> hist = metrics_histogram("size")
>>> hist.observe(100)
>>> hist.observe(200)
>>> hist.count
2

percentile ¶

percentile(p: float) -> float

Calculate percentile.

Parameters:

Name	Type	Description	Default
`p`	`float`	Percentile (0-100).	required

Returns:

Type	Description
`float`	Value at percentile.

Examples:

>>> hist = metrics_histogram("values")
>>> for i in range(100):
...     hist.observe(i)
>>> hist.percentile(50)
49.5

reset ¶

reset() -> None

Clear all observations.

metrics_timer ¶

Timer Metric¶

Timer to measure durations as context manager.

Examples¶

from rite.diagnostics.metrics import metrics_timer timer = metrics_timer("operation") with timer: ... pass timer.count 1

Classes¶

metrics_timer ¶

metrics_timer(name: str)

Timer metric using context manager.

Attributes:

Name	Type	Description
`name`		Metric name.
`count`	`int`	Number of timing measurements.
`total`	`float`	Total time measured.

Examples:

>>> timer = metrics_timer("request")
>>> with timer:
...     time.sleep(0.01)
>>> timer.count
1
>>> timer.total > 0
True

Initialize timer.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

average property ¶

average: float

Get average time per measurement.

count property ¶

count: int

Get number of measurements.

total property ¶

total: float

Get total time in seconds.

Functions¶

reset ¶

reset() -> None

Reset timer statistics.

metrics_counter ¶

Counter Metric¶

Counter that can only increase.

Examples¶

from rite.diagnostics.metrics import metrics_counter counter = metrics_counter("requests") counter.increment() counter.value 1

Classes¶

metrics_counter ¶

metrics_counter(name: str)

Counter metric that only increases.

Attributes:

Name	Type	Description
`name`		Metric name.
`value`	`float`	Current counter value.

Examples:

>>> counter = metrics_counter("api_calls")
>>> counter.increment()
>>> counter.increment(5)
>>> counter.value
6

Initialize counter.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

value property ¶

value: float

Get current value.

Functions¶

increment ¶

increment(amount: float = 1.0) -> None

Increment counter.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to add (must be non-negative).	`1.0`

Raises:

Type	Description
`ValueError`	If amount is negative.

Examples:

>>> counter = metrics_counter("requests")
>>> counter.increment()
>>> counter.increment(10)
>>> counter.value
11.0

reset ¶

reset() -> None

Reset counter to zero.

metrics_gauge ¶

Gauge Metric¶

Gauge that can increase or decrease.

Examples¶

from rite.diagnostics.metrics import metrics_gauge gauge = metrics_gauge("temperature") gauge.set(25.0) gauge.value 25.0

Classes¶

metrics_gauge ¶

metrics_gauge(name: str, initial_value: float = 0.0)

Gauge metric that can go up or down.

Attributes:

Name	Type	Description
`name`		Metric name.
`value`	`float`	Current gauge value.

Examples:

>>> gauge = metrics_gauge("memory_usage")
>>> gauge.set(100)
>>> gauge.increment(50)
>>> gauge.decrement(25)
>>> gauge.value
125.0

Initialize gauge.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required
`initial_value`	`float`	Starting value.	`0.0`

Attributes¶

value property ¶

value: float

Get current value.

Functions¶

decrement ¶

decrement(amount: float = 1.0) -> None

Decrease gauge value.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to subtract.	`1.0`

increment ¶

increment(amount: float = 1.0) -> None

Increase gauge value.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to add.	`1.0`

reset ¶

reset() -> None

Reset gauge to zero.

set ¶

set(value: float) -> None

Set gauge to specific value.

Parameters:

Name	Type	Description	Default
`value`	`float`	New value.	required

Examples:

>>> gauge = metrics_gauge("cpu")
>>> gauge.set(75.5)
>>> gauge.value
75.5

metrics_histogram ¶

Histogram Metric¶

Histogram to track distribution of values.

Examples¶

from rite.diagnostics.metrics import metrics_histogram hist = metrics_histogram("response_time") hist.observe(0.5) hist.count 1

Classes¶

metrics_histogram ¶

metrics_histogram(name: str)

Histogram metric for value distributions.

Attributes:

Name	Type	Description
`name`		Metric name.
`count`	`int`	Number of observations.
`sum`	`float`	Sum of all values.

Examples:

>>> hist = metrics_histogram("latency")
>>> hist.observe(1.0)
>>> hist.observe(2.0)
>>> hist.observe(3.0)
>>> hist.mean
2.0

Initialize histogram.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

count property ¶

count: int

Get number of observations.

mean property ¶

mean: float

Get mean of values.

median property ¶

median: float

Get median of values.

sum property ¶

sum: float

Get sum of all values.

Functions¶

observe ¶

observe(value: float) -> None

Record observation.

Parameters:

Name	Type	Description	Default
`value`	`float`	Value to observe.	required

Examples:

>>> hist = metrics_histogram("size")
>>> hist.observe(100)
>>> hist.observe(200)
>>> hist.count
2

percentile ¶

percentile(p: float) -> float

Calculate percentile.

Parameters:

Name	Type	Description	Default
`p`	`float`	Percentile (0-100).	required

Returns:

Type	Description
`float`	Value at percentile.

Examples:

>>> hist = metrics_histogram("values")
>>> for i in range(100):
...     hist.observe(i)
>>> hist.percentile(50)
49.5

reset ¶

reset() -> None

Clear all observations.

metrics_timer ¶

Timer Metric¶

Timer to measure durations as context manager.

Examples¶

from rite.diagnostics.metrics import metrics_timer timer = metrics_timer("operation") with timer: ... pass timer.count 1

Classes¶

metrics_timer ¶

metrics_timer(name: str)

Timer metric using context manager.

Attributes:

Name	Type	Description
`name`		Metric name.
`count`	`int`	Number of timing measurements.
`total`	`float`	Total time measured.

Examples:

>>> timer = metrics_timer("request")
>>> with timer:
...     time.sleep(0.01)
>>> timer.count
1
>>> timer.total > 0
True

Initialize timer.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

average property ¶

average: float

Get average time per measurement.

count property ¶

count: int

Get number of measurements.

total property ¶

total: float

Get total time in seconds.

Functions¶

reset ¶

reset() -> None

Reset timer statistics.

profiling ¶

Profiling Module¶

Performance profiling utilities.

This submodule provides utilities for measuring execution time, memory usage, and function call counts.

Examples¶

from rite.diagnostics.profiling import ( ... profiling_timer, ... profiling_stopwatch, ... profiling_memory ... ) @profiling_timer() ... def slow_function(): ... pass with profiling_stopwatch("task") as sw: ... pass

Modules¶

profiling_count_calls ¶

Function Call Counter¶

Count how many times a function is called.

Examples¶

from rite.diagnostics.profiling import profiling_count_calls @profiling_count_calls() ... def api_call(): ... pass

Functions¶

profiling_count_calls ¶

profiling_count_calls(print_every: int | None = None) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to count function calls.

Parameters:

Name	Type	Description	Default
`print_every`	`int \| None`	Print count every N calls (None = don't print).	`None`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with call counter.

Examples:

>>> @profiling_count_calls()
... def process():
...     pass
>>> process()
>>> process.call_count
1
>>> @profiling_count_calls(print_every=10)
... def api_request():
...     pass

profiling_memory ¶

Memory Profiler¶

Measure memory usage of function.

Examples¶

from rite.diagnostics.profiling import profiling_memory @profiling_memory() ... def memory_intensive(): ... data = [0] * 1000000

Functions¶

profiling_memory ¶

profiling_memory(print_result: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to measure memory usage of function.

Parameters:

Name	Type	Description	Default
`print_result`	`bool`	If True, print memory usage.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function that measures memory.

Examples:

>>> @profiling_memory()
... def create_large_list():
...     return [0] * 1000000
>>> @profiling_memory(print_result=False)
... def process_data():
...     pass

Notes

Uses sys.getsizeof for basic measurement. For detailed profiling, use memory_profiler package.

profiling_stopwatch ¶

Timer Context Manager¶

Context manager to measure elapsed time.

Examples¶

from rite.diagnostics.profiling import profiling_stopwatch with profiling_stopwatch("operation") as sw: ... time.sleep(0.1) print(sw.elapsed)

Classes¶

profiling_stopwatch ¶

profiling_stopwatch(name: str = 'operation')

Context manager to measure elapsed time.

Attributes:

Name	Type	Description
`name`		Operation name.
`elapsed`	`float`	Elapsed time in seconds.
`start_time`	`float`	Start timestamp.
`end_time`	`float`	End timestamp.

Examples:

>>> with profiling_stopwatch("task") as sw:
...     time.sleep(0.1)
>>> sw.elapsed > 0.1
True

Initialize stopwatch.

Parameters:

Name	Type	Description	Default
`name`	`str`	Operation name for display.	`'operation'`

profiling_timer ¶

Timer Decorator¶

Decorator to measure function execution time.

Examples¶

from rite.diagnostics.profiling import profiling_timer @profiling_timer() ... def slow_function(): ... time.sleep(1)

Functions¶

profiling_timer ¶

profiling_timer(name: str | None = None, print_result: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to measure function execution time.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Optional name for the operation being timed.	`None`
`print_result`	`bool`	If True, print timing result.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function that measures execution time.

Examples:

>>> @profiling_timer()
... def compute():
...     sum(range(1000000))
>>> @profiling_timer(name="API Call", print_result=False)
... def api_request():
...     pass

profiling_count_calls ¶

Function Call Counter¶

Count how many times a function is called.

Examples¶

from rite.diagnostics.profiling import profiling_count_calls @profiling_count_calls() ... def api_call(): ... pass

Functions¶

profiling_count_calls ¶

profiling_count_calls(print_every: int | None = None) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to count function calls.

Parameters:

Name	Type	Description	Default
`print_every`	`int \| None`	Print count every N calls (None = don't print).	`None`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with call counter.

Examples:

>>> @profiling_count_calls()
... def process():
...     pass
>>> process()
>>> process.call_count
1
>>> @profiling_count_calls(print_every=10)
... def api_request():
...     pass

profiling_memory ¶

Memory Profiler¶

Measure memory usage of function.

Examples¶

from rite.diagnostics.profiling import profiling_memory @profiling_memory() ... def memory_intensive(): ... data = [0] * 1000000

Functions¶

profiling_memory ¶

profiling_memory(print_result: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to measure memory usage of function.

Parameters:

Name	Type	Description	Default
`print_result`	`bool`	If True, print memory usage.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function that measures memory.

Examples:

>>> @profiling_memory()
... def create_large_list():
...     return [0] * 1000000
>>> @profiling_memory(print_result=False)
... def process_data():
...     pass

Notes

Uses sys.getsizeof for basic measurement. For detailed profiling, use memory_profiler package.

profiling_stopwatch ¶

Timer Context Manager¶

Context manager to measure elapsed time.

Examples¶

from rite.diagnostics.profiling import profiling_stopwatch with profiling_stopwatch("operation") as sw: ... time.sleep(0.1) print(sw.elapsed)

Classes¶

profiling_stopwatch ¶

profiling_stopwatch(name: str = 'operation')

Context manager to measure elapsed time.

Attributes:

Name	Type	Description
`name`		Operation name.
`elapsed`	`float`	Elapsed time in seconds.
`start_time`	`float`	Start timestamp.
`end_time`	`float`	End timestamp.

Examples:

>>> with profiling_stopwatch("task") as sw:
...     time.sleep(0.1)
>>> sw.elapsed > 0.1
True

Initialize stopwatch.

Parameters:

Name	Type	Description	Default
`name`	`str`	Operation name for display.	`'operation'`

profiling_timer ¶

Timer Decorator¶

Decorator to measure function execution time.

Examples¶

from rite.diagnostics.profiling import profiling_timer @profiling_timer() ... def slow_function(): ... time.sleep(1)

Functions¶

profiling_timer ¶

profiling_timer(name: str | None = None, print_result: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to measure function execution time.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Optional name for the operation being timed.	`None`
`print_result`	`bool`	If True, print timing result.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function that measures execution time.

Examples:

>>> @profiling_timer()
... def compute():
...     sum(range(1000000))
>>> @profiling_timer(name="API Call", print_result=False)
... def api_request():
...     pass

Submodules¶

Logging¶

Structured logging utilities.

Logging Module¶

Logging utilities with file, console, structured, and contextual output.

This submodule provides various logging configurations for different use cases: file logging with rotation, console logging with colors, structured JSON logging, and context-aware logging.

Examples¶

from rite.diagnostics.logging import ( ... logging_to_file, ... logging_to_console, ... logging_structured ... ) file_logger = logging_to_file("app", "app.log") console_logger = logging_to_console("debug", colorize=True) struct_logger = logging_structured("api")

Modules¶

logging_to_console ¶

Console Logger¶

Create logger that writes to console/stdout.

Examples¶

from rite.diagnostics.logging import logging_to_console logger = logging_to_console("app") logger.info("Application started")

Functions¶

logging_to_console ¶

logging_to_console(name: str, level: int = logging.INFO, format_string: str | None = None, colorize: bool = False) -> logging.Logger

Create logger that writes to console.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`format_string`	`str \| None`	Custom format string.	`None`
`colorize`	`bool`	If True, colorize output by level.	`False`

Returns:

Type	Description
`Logger`	Configured console logger.

Examples:

>>> logger = logging_to_console("app")
>>> logger.info("Hello world")
>>> logger = logging_to_console("debug", level=logging.DEBUG)
>>> logger.debug("Debug message")

logging_to_file ¶

File Logger¶

Create logger that writes to a file with rotation support.

Examples¶

from rite.diagnostics.logging import logging_to_file logger = logging_to_file("app", "app.log") logger.info("Application started")

Functions¶

logging_to_file ¶

logging_to_file(name: str, filepath: str, level: int = logging.INFO, max_bytes: int = 10485760, backup_count: int = 5, format_string: str | None = None) -> logging.Logger

Create logger that writes to a file with rotation.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`filepath`	`str`	Path to log file.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`max_bytes`	`int`	Max file size before rotation (default 10MB).	`10485760`
`backup_count`	`int`	Number of backup files to keep (default 5).	`5`
`format_string`	`str \| None`	Custom format string.	`None`

Returns:

Type	Description
`Logger`	Configured file logger.

Examples:

>>> logger = logging_to_file("app", "logs/app.log")
>>> logger.info("Starting application")
>>> logger = logging_to_file(
...     "api",
...     "logs/api.log",
...     max_bytes=5242880,
...     backup_count=3
... )

logging_structured ¶

Structured Logger¶

Logger with structured output (JSON, key-value pairs).

Examples¶

from rite.diagnostics.logging import logging_structured logger = logging_structured("app") logger.info("User logged in", user_id=123, ip="192.168.1.1")

Functions¶

logging_structured ¶

logging_structured(name: str, level: int = logging.INFO, json_format: bool = True) -> logging.Logger

Create logger with structured output.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`json_format`	`bool`	If True, output as JSON; else key=value format.	`True`

Returns:

Type	Description
`Logger`	Configured logger with structured formatter.

Examples:

>>> logger = logging_structured("app")
>>> logger.info("Event occurred", user_id=123)
>>> logger = logging_structured("api", json_format=False)
>>> logger.warning("Slow query", duration_ms=500)

logging_with_context ¶

Context Logger¶

Logger that includes contextual information in all messages.

Examples¶

from rite.diagnostics.logging import logging_with_context logger = logging_with_context("app", request_id="abc123") logger.info("Processing request")

Functions¶

logging_with_context ¶

logging_with_context(name: str, level: int = logging.INFO, **context: Any) -> logging.LoggerAdapter

Create logger that includes context in all messages.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name.	required
`level`	`int`	Logging level (default INFO).	`INFO`
`**context`	`Any`	Context key-value pairs to include.	`{}`

Returns:

Type	Description
`LoggerAdapter`	Logger adapter with context.

Examples:

>>> logger = logging_with_context(
...     "api",
...     request_id="req123",
...     user_id=456
... )
>>> logger.info("Request processed")
>>> logger = logging_with_context(
...     "worker",
...     worker_id="w1",
...     queue="default"
... )

Profiling¶

Performance profiling tools.

Profiling Module¶

Performance profiling utilities.

This submodule provides utilities for measuring execution time, memory usage, and function call counts.

Examples¶

from rite.diagnostics.profiling import ( ... profiling_timer, ... profiling_stopwatch, ... profiling_memory ... ) @profiling_timer() ... def slow_function(): ... pass with profiling_stopwatch("task") as sw: ... pass

Modules¶

profiling_timer ¶

Timer Decorator¶

Decorator to measure function execution time.

Examples¶

from rite.diagnostics.profiling import profiling_timer @profiling_timer() ... def slow_function(): ... time.sleep(1)

Functions¶

profiling_timer ¶

profiling_timer(name: str | None = None, print_result: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to measure function execution time.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Optional name for the operation being timed.	`None`
`print_result`	`bool`	If True, print timing result.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function that measures execution time.

Examples:

>>> @profiling_timer()
... def compute():
...     sum(range(1000000))
>>> @profiling_timer(name="API Call", print_result=False)
... def api_request():
...     pass

profiling_stopwatch ¶

Timer Context Manager¶

Context manager to measure elapsed time.

Examples¶

from rite.diagnostics.profiling import profiling_stopwatch with profiling_stopwatch("operation") as sw: ... time.sleep(0.1) print(sw.elapsed)

Classes¶

profiling_stopwatch ¶

profiling_stopwatch(name: str = 'operation')

Context manager to measure elapsed time.

Attributes:

Name	Type	Description
`name`		Operation name.
`elapsed`	`float`	Elapsed time in seconds.
`start_time`	`float`	Start timestamp.
`end_time`	`float`	End timestamp.

Examples:

>>> with profiling_stopwatch("task") as sw:
...     time.sleep(0.1)
>>> sw.elapsed > 0.1
True

Initialize stopwatch.

Parameters:

Name	Type	Description	Default
`name`	`str`	Operation name for display.	`'operation'`

profiling_memory ¶

Memory Profiler¶

Measure memory usage of function.

Examples¶

from rite.diagnostics.profiling import profiling_memory @profiling_memory() ... def memory_intensive(): ... data = [0] * 1000000

Functions¶

profiling_memory ¶

profiling_memory(print_result: bool = True) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to measure memory usage of function.

Parameters:

Name	Type	Description	Default
`print_result`	`bool`	If True, print memory usage.	`True`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function that measures memory.

Examples:

>>> @profiling_memory()
... def create_large_list():
...     return [0] * 1000000
>>> @profiling_memory(print_result=False)
... def process_data():
...     pass

Notes

Uses sys.getsizeof for basic measurement. For detailed profiling, use memory_profiler package.

profiling_count_calls ¶

Function Call Counter¶

Count how many times a function is called.

Examples¶

from rite.diagnostics.profiling import profiling_count_calls @profiling_count_calls() ... def api_call(): ... pass

Functions¶

profiling_count_calls ¶

profiling_count_calls(print_every: int | None = None) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to count function calls.

Parameters:

Name	Type	Description	Default
`print_every`	`int \| None`	Print count every N calls (None = don't print).	`None`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with call counter.

Examples:

>>> @profiling_count_calls()
... def process():
...     pass
>>> process()
>>> process.call_count
1
>>> @profiling_count_calls(print_every=10)
... def api_request():
...     pass

Error Handling¶

Error management utilities.

Error Handling Module¶

Error handling utilities with retry, catching, and formatting.

This submodule provides utilities for handling errors gracefully: retry decorators, exception context managers, traceback formatting, and exception chain analysis.

Examples¶

from rite.diagnostics.errors import errors_retry, errors_catch @errors_retry(max_attempts=3) ... def flaky_function(): ... pass with errors_catch(): ... risky_operation()

Modules¶

errors_catch ¶

Exception Context Manager¶

Context manager to catch and handle exceptions gracefully.

Examples¶

from rite.diagnostics.errors import errors_catch with errors_catch(): ... risky_operation()

Functions¶

errors_catch ¶

errors_catch(exceptions: tuple[type[Exception], ...] = (Exception,), handler: Callable[[Exception], Any] | None = None, reraise: bool = False, logger: Logger | None = None) -> Generator[None, None, None]

Context manager to catch and handle exceptions.

Parameters:

Name	Type	Description	Default
`exceptions`	`tuple[type[Exception], ...]`	Tuple of exceptions to catch.	`(Exception,)`
`handler`	`Callable[[Exception], Any] \| None`	Optional handler function called with exception.	`None`
`reraise`	`bool`	If True, reraise exception after handling.	`False`
`logger`	`Logger \| None`	Optional logger to log exceptions.	`None`

Yields:

Type	Description
`None`	None.

Examples:

>>> with errors_catch():
...     1 / 0
>>> with errors_catch(handler=lambda e: print(f"Error: {e}")):
...     risky_operation()
>>> with errors_catch((ValueError, KeyError), reraise=True):
...     parse_data()

errors_retry ¶

Retry Decorator¶

Decorator to retry function calls on failure.

Examples¶

from rite.diagnostics.errors import errors_retry @errors_retry(max_attempts=3, delay=1.0) ... def unstable_function(): ... pass

Functions¶

errors_retry ¶

errors_retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0, exceptions: tuple[type[Exception], ...] = (Exception,)) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorator to retry function on failure.

Parameters:

Name	Type	Description	Default
`max_attempts`	`int`	Maximum retry attempts.	`3`
`delay`	`float`	Initial delay between retries (seconds).	`1.0`
`backoff`	`float`	Multiplier for delay on each retry.	`2.0`
`exceptions`	`tuple[type[Exception], ...]`	Tuple of exceptions to catch.	`(Exception,)`

Returns:

Type	Description
`Callable[[Callable[..., T]], Callable[..., T]]`	Decorated function with retry logic.

Examples:

>>> @errors_retry(max_attempts=3)
... def flaky_api_call():
...     pass
>>> @errors_retry(max_attempts=5, delay=0.5, backoff=1.5)
... def database_query():
...     pass

errors_get_chain ¶

Exception Chain¶

Get exception chain (cause hierarchy).

Examples¶

from rite.diagnostics.errors import errors_get_chain try: ... try: ... 1 / 0 ... except ZeroDivisionError as e: ... raise ValueError("Wrapped") from e ... except ValueError as e: ... chain = errors_get_chain(e)

Functions¶

errors_get_chain ¶

errors_get_chain(exception: BaseException) -> list[BaseException]

Get exception chain (cause hierarchy).

Parameters:

Name	Type	Description	Default
`exception`	`BaseException`	Exception to get chain for.	required

Returns:

Type	Description
`list[BaseException]`	List of exceptions from root cause to final exception.

Examples:

>>> try:
...     try:
...         raise ValueError("root")
...     except ValueError as e:
...         raise KeyError("wrapped") from e
... except KeyError as e:
...     chain = errors_get_chain(e)
...     len(chain)
2

errors_format_traceback ¶

Traceback Formatter¶

Format exception tracebacks for better readability.

Examples¶

from rite.diagnostics.errors import errors_format_traceback try: ... 1 / 0 ... except Exception as e: ... print(errors_format_traceback(e))

Functions¶

errors_format_traceback ¶

errors_format_traceback(exception: BaseException, include_locals: bool = False) -> str

Format exception traceback as string.

Parameters:

Name	Type	Description	Default
`exception`	`BaseException`	Exception to format.	required
`include_locals`	`bool`	If True, include local variables.	`False`

Returns:

Type	Description
`str`	Formatted traceback string.

Examples:

>>> try:
...     raise ValueError("test error")
... except ValueError as e:
...     tb = errors_format_traceback(e)
...     "ValueError" in tb
True

Metrics¶

Application metrics collection.

Metrics Module¶

Performance and monitoring metrics.

This submodule provides metric classes for counters, gauges, histograms, and timers.

Examples¶

from rite.diagnostics.metrics import ( ... metrics_counter, ... metrics_gauge, ... metrics_histogram, ... metrics_timer ... ) counter = metrics_counter("requests") counter.increment()

Modules¶

metrics_counter ¶

Counter Metric¶

Counter that can only increase.

Examples¶

from rite.diagnostics.metrics import metrics_counter counter = metrics_counter("requests") counter.increment() counter.value 1

Classes¶

metrics_counter ¶

metrics_counter(name: str)

Counter metric that only increases.

Attributes:

Name	Type	Description
`name`		Metric name.
`value`	`float`	Current counter value.

Examples:

>>> counter = metrics_counter("api_calls")
>>> counter.increment()
>>> counter.increment(5)
>>> counter.value
6

Initialize counter.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

value property ¶

value: float

Get current value.

Functions¶

increment ¶

increment(amount: float = 1.0) -> None

Increment counter.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to add (must be non-negative).	`1.0`

Raises:

Type	Description
`ValueError`	If amount is negative.

Examples:

>>> counter = metrics_counter("requests")
>>> counter.increment()
>>> counter.increment(10)
>>> counter.value
11.0

reset ¶

reset() -> None

Reset counter to zero.

metrics_gauge ¶

Gauge Metric¶

Gauge that can increase or decrease.

Examples¶

from rite.diagnostics.metrics import metrics_gauge gauge = metrics_gauge("temperature") gauge.set(25.0) gauge.value 25.0

Classes¶

metrics_gauge ¶

metrics_gauge(name: str, initial_value: float = 0.0)

Gauge metric that can go up or down.

Attributes:

Name	Type	Description
`name`		Metric name.
`value`	`float`	Current gauge value.

Examples:

>>> gauge = metrics_gauge("memory_usage")
>>> gauge.set(100)
>>> gauge.increment(50)
>>> gauge.decrement(25)
>>> gauge.value
125.0

Initialize gauge.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required
`initial_value`	`float`	Starting value.	`0.0`

Attributes¶

value property ¶

value: float

Get current value.

Functions¶

decrement ¶

decrement(amount: float = 1.0) -> None

Decrease gauge value.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to subtract.	`1.0`

increment ¶

increment(amount: float = 1.0) -> None

Increase gauge value.

Parameters:

Name	Type	Description	Default
`amount`	`float`	Amount to add.	`1.0`

reset ¶

reset() -> None

Reset gauge to zero.

set ¶

set(value: float) -> None

Set gauge to specific value.

Parameters:

Name	Type	Description	Default
`value`	`float`	New value.	required

Examples:

>>> gauge = metrics_gauge("cpu")
>>> gauge.set(75.5)
>>> gauge.value
75.5

metrics_timer ¶

Timer Metric¶

Timer to measure durations as context manager.

Examples¶

from rite.diagnostics.metrics import metrics_timer timer = metrics_timer("operation") with timer: ... pass timer.count 1

Classes¶

metrics_timer ¶

metrics_timer(name: str)

Timer metric using context manager.

Attributes:

Name	Type	Description
`name`		Metric name.
`count`	`int`	Number of timing measurements.
`total`	`float`	Total time measured.

Examples:

>>> timer = metrics_timer("request")
>>> with timer:
...     time.sleep(0.01)
>>> timer.count
1
>>> timer.total > 0
True

Initialize timer.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

average property ¶

average: float

Get average time per measurement.

count property ¶

count: int

Get number of measurements.

total property ¶

total: float

Get total time in seconds.

Functions¶

reset ¶

reset() -> None

Reset timer statistics.

metrics_histogram ¶

Histogram Metric¶

Histogram to track distribution of values.

Examples¶

from rite.diagnostics.metrics import metrics_histogram hist = metrics_histogram("response_time") hist.observe(0.5) hist.count 1

Classes¶

metrics_histogram ¶

metrics_histogram(name: str)

Histogram metric for value distributions.

Attributes:

Name	Type	Description
`name`		Metric name.
`count`	`int`	Number of observations.
`sum`	`float`	Sum of all values.

Examples:

>>> hist = metrics_histogram("latency")
>>> hist.observe(1.0)
>>> hist.observe(2.0)
>>> hist.observe(3.0)
>>> hist.mean
2.0

Initialize histogram.

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name.	required

Attributes¶

count property ¶

count: int

Get number of observations.

mean property ¶

mean: float

Get mean of values.

median property ¶

median: float

Get median of values.

sum property ¶

sum: float

Get sum of all values.

Functions¶

observe ¶

observe(value: float) -> None

Record observation.

Parameters:

Name	Type	Description	Default
`value`	`float`	Value to observe.	required

Examples:

>>> hist = metrics_histogram("size")
>>> hist.observe(100)
>>> hist.observe(200)
>>> hist.count
2

percentile ¶

percentile(p: float) -> float

Calculate percentile.

Parameters:

Name	Type	Description	Default
`p`	`float`	Percentile (0-100).	required

Returns:

Type	Description
`float`	Value at percentile.

Examples:

>>> hist = metrics_histogram("values")
>>> for i in range(100):
...     hist.observe(i)
>>> hist.percentile(50)
49.5

reset ¶

reset() -> None

Clear all observations.

Examples¶

from rite.diagnostics import (
    profiling_timer,
    errors_retry,
    metrics_counter
)

# Time function execution
@profiling_timer
def slow_function():
    # ... code ...
    pass

# Retry on failure
@errors_retry(max_attempts=3)
def unstable_operation():
    # ... code ...
    pass

# Count metrics
counter = metrics_counter("requests")
counter.increment()