pynenc.util.redis_debug_client

Redis Debug Client

This module provides instrumentation for Redis commands to track performance and identify bottlenecks. It captures detailed call stacks, timing information, and aggregates statistics to help diagnose Redis performance issues.

Module Contents

Classes

CommandStats	Statistics for a Redis command.
CallStack	Information about a call stack.
CallerInfo	Information about a Redis command caller.
RedisStats	Tracks Redis command statistics with call stack information.
DebugRedisClient	Redis client that logs all commands with timing information.

Functions

get_frame_info	Extract caller information from a frame.
collect_call_stack	Collect call stack information from the current execution context.
format_arg	Format a single argument for logging, truncating if too long.
format_caller_for_display	Format a caller string for display, with clean truncation for better alignment.
format_truncated_caller	Format a long caller string with clean truncation.
_format_long_caller	Format a long caller with many path components.
_format_short_caller	Format a shorter caller with fewer path components.
log_redis_command	Decorator to log Redis commands with timing and call stack information.
patch_redis_client	Patch the get_redis_client function to return a DebugRedisClient instance. This ensures all Redis operations throughout the application are monitored.
log_pool_stats	Log statistics about all Redis connection pools.
monitor_redis_pools	Start a background thread to monitor Redis connection pools.
start_redis_debugging	Start Redis debugging with optional pool monitoring.
start_tracking_redis_stats	Start tracking Redis command statistics.
stop_tracking_redis_stats	Stop tracking Redis command statistics.
get_redis_stats_report	Get a formatted report of Redis command statistics.
get_sorted_redis_commands	Get Redis commands sorted by total execution time.
deep_patch_redis	Patch Redis at a deeper level to guarantee all Redis operations are tracked. This patches the Redis module’s core methods that all clients use.

Data

logger
T
DEFAULT_STACK_DEPTH
MAX_CALL_STACKS_PER_CALLER
MAX_TOP_CALLERS
SLOW_COMMAND_THRESHOLD
MAX_ARG_LENGTH
redis_stats

API

pynenc.util.redis_debug_client.logger

‘getLogger(…)’

pynenc.util.redis_debug_client.T

‘TypeVar(…)’

pynenc.util.redis_debug_client.DEFAULT_STACK_DEPTH

7

pynenc.util.redis_debug_client.MAX_CALL_STACKS_PER_CALLER

5

pynenc.util.redis_debug_client.MAX_TOP_CALLERS

10

pynenc.util.redis_debug_client.SLOW_COMMAND_THRESHOLD

0.1

pynenc.util.redis_debug_client.MAX_ARG_LENGTH

100

class pynenc.util.redis_debug_client.CommandStats

Bases: typing.NamedTuple

Statistics for a Redis command.

count: int

None

total_time: float

None

avg_time: float

None

max_time: float

None

class pynenc.util.redis_debug_client.CallStack

Bases: typing.NamedTuple

Information about a call stack.

frames: list[str]

None

count: int

1

class pynenc.util.redis_debug_client.CallerInfo

Bases: typing.NamedTuple

Information about a Redis command caller.

module_name: str

None

class_name: Optional[str]

None

function_name: str

None

line_number: int

None

property formatted_caller: str

Format the caller information as a string.

pynenc.util.redis_debug_client.get_frame_info(frame: inspect.FrameInfo) → pynenc.util.redis_debug_client.CallerInfo

Extract caller information from a frame.

Parameters:

frame – Stack frame to analyze

Returns:

Structured caller information

pynenc.util.redis_debug_client.collect_call_stack() → tuple[str, list[str]]

Collect call stack information from the current execution context.

Returns:

A tuple containing (primary_caller, call_stack)

pynenc.util.redis_debug_client.format_arg(arg: Any) → str

Format a single argument for logging, truncating if too long.

Parameters:

arg – Argument to format

Returns:

String representation of the argument

pynenc.util.redis_debug_client.format_caller_for_display(caller: str, width: int) → str

Format a caller string for display, with clean truncation for better alignment.

Parameters:

• caller – The caller string to format

• width – Maximum width for display

Returns:

Formatted caller string with consistent width

pynenc.util.redis_debug_client.format_truncated_caller(caller: str, width: int) → str

Format a long caller string with clean truncation.

Uses a consistent pattern to ensure table alignment:

• For long paths: keeps start and end with … in middle

• Guarantees exact width output

Parameters:

• caller – Caller identifier to format

• width – Exact width to format to

Returns:

Formatted caller string with exact width

pynenc.util.redis_debug_client._format_long_caller(caller_parts: list[str], width: int) → str

Format a long caller with many path components.

pynenc.util.redis_debug_client._format_short_caller(caller: str, caller_parts: list[str], width: int) → str

Format a shorter caller with fewer path components.

class pynenc.util.redis_debug_client.RedisStats

Tracks Redis command statistics with call stack information.

Initialization

Initialize Redis stats tracking.

start_tracking() → None

Start tracking Redis command statistics.

stop_tracking() → None

Stop tracking Redis command statistics.

add_stat(caller: str, call_stack: list[str], command: str, execution_time: float) → None

Add a command execution stat with call stack information.

Parameters:

• caller – The code location that called the Redis command

• call_stack – The call frames leading to this command

• command – The Redis command name

• execution_time – How long the command took to execute

_update_command_stats(caller: str, command: str, execution_time: float) → None

Update statistics for a specific command.

_update_call_stack(caller: str, call_stack: list[str]) → None

Update call stack information for a caller.

get_sorted_commands() → list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]]

Get commands sorted by total time (for programmatic use).

get_report(top_n: int = 20, stack_depth: int = DEFAULT_STACK_DEPTH, table_width: int = 140) → str

Generate a performance report for Redis commands.

Parameters:

• top_n – Number of top commands to include

• stack_depth – Number of frames to show in call stacks

• table_width – Maximum width for the table

Returns:

Formatted report as a string

_get_sorted_commands_list() → list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]]

Get commands sorted by total execution time.

_get_summary_stats(all_commands: list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]]) → list[str]

Generate the summary statistics section.

_get_top_callers_section(all_commands: list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]], stack_depth: int) → list[str]

Generate the top callers section with call stacks and improved formatting.

_is_duplicate_path(new_path: tuple[str, ...], existing_paths: set[tuple[str, ...]]) → bool

Check if a call path is essentially a duplicate of an existing path.

This compares the function names and classes, ignoring minor line number differences.

Parameters:

• new_path – The new call path to check

• existing_paths – Previously shown call paths

Returns:

True if this path is essentially a duplicate

_format_frame_for_display(frame: str) → str

Format a stack frame for display.

Highlights important parts of the frame and improves readability.

Parameters:

frame – Raw stack frame string

Returns:

Formatted frame for display

_format_commands_table(all_commands: list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]], top_n: int, table_width: int = 140) → list[str]

Format the commands table section of the report with improved alignment.

_aggregate_stats_by_caller(all_commands: list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]]) → dict[str, pynenc.util.redis_debug_client.CommandStats]

Aggregate command statistics by caller.

pynenc.util.redis_debug_client.log_redis_command(func: Callable[..., pynenc.util.redis_debug_client.T]) → Callable[..., pynenc.util.redis_debug_client.T]

Decorator to log Redis commands with timing and call stack information.

Parameters:

func – Redis command function to wrap

Returns:

Wrapped function with logging and stats collection

class pynenc.util.redis_debug_client.DebugRedisClient(host='localhost', port=6379, db=0, password=None, socket_timeout=None, socket_connect_timeout=None, socket_keepalive=None, socket_keepalive_options=None, connection_pool=None, unix_socket_path=None, encoding='utf-8', encoding_errors='strict', charset=None, errors=None, decode_responses=False, retry_on_timeout=False, retry_on_error=None, ssl=False, ssl_keyfile=None, ssl_certfile=None, ssl_cert_reqs='required', ssl_ca_certs=None, ssl_ca_path=None, ssl_ca_data=None, ssl_check_hostname=False, ssl_password=None, ssl_validate_ocsp=False, ssl_validate_ocsp_stapled=False, ssl_ocsp_context=None, ssl_ocsp_expected_cert=None, ssl_min_version=None, ssl_ciphers=None, max_connections=None, single_connection_client=False, health_check_interval=0, client_name=None, lib_name='redis-py', lib_version=get_lib_version(), username=None, retry=None, redis_connect_func=None, credential_provider: Optional[redis.credentials.CredentialProvider] = None, protocol: Optional[int] = 2, cache: Optional[redis.cache.CacheInterface] = None, cache_config: Optional[redis.cache.CacheConfig] = None)

Bases: redis.Redis

Redis client that logs all commands with timing information.

Initialization

Initialize a new Redis client. To specify a retry policy for specific errors, first set retry_on_error to a list of the error/s to retry on, then set retry to a valid Retry object. To retry on TimeoutError, retry_on_timeout can also be set to True.

Args:

single_connection_client: if True, connection pool is not used. In that case Redis instance use is not thread safe.

__getattribute__(name: str) → Any

Intercept attribute access to wrap Redis commands with logging.

Parameters:

name – Attribute name

Returns:

Wrapped command or original attribute

_execute_command(*args: Any, **options: Any) → Any

Override the core Redis client execute_command method to track all operations.

This is the central method that all Redis commands ultimately call.

pynenc.util.redis_debug_client.patch_redis_client() → None

Patch the get_redis_client function to return a DebugRedisClient instance. This ensures all Redis operations throughout the application are monitored.

pynenc.util.redis_debug_client.log_pool_stats() → None

Log statistics about all Redis connection pools.

pynenc.util.redis_debug_client.monitor_redis_pools(interval_seconds: int = 60) → None

Start a background thread to monitor Redis connection pools.

Parameters:

interval_seconds – How often to log pool statistics

pynenc.util.redis_debug_client.start_redis_debugging(monitor_pools: bool = True, monitor_interval: int = 60) → None

Start Redis debugging with optional pool monitoring.

Parameters:

• monitor_pools – Whether to monitor connection pools

• monitor_interval – How often to log pool statistics (seconds)

pynenc.util.redis_debug_client.redis_stats

‘RedisStats(…)’

pynenc.util.redis_debug_client.start_tracking_redis_stats() → None

Start tracking Redis command statistics.

pynenc.util.redis_debug_client.stop_tracking_redis_stats() → None

Stop tracking Redis command statistics.

pynenc.util.redis_debug_client.get_redis_stats_report(top_n: int = 20, stack_depth: int = DEFAULT_STACK_DEPTH, table_width: int = 140) → str

Get a formatted report of Redis command statistics.

Parameters:

• top_n – Number of top commands to include in the report

• stack_depth – Number of call stack frames to show for each caller

• table_width – Maximum width for the table display

Returns:

Formatted report

pynenc.util.redis_debug_client.get_sorted_redis_commands() → list[tuple[str, str, pynenc.util.redis_debug_client.CommandStats]]

Get Redis commands sorted by total execution time.

pynenc.util.redis_debug_client.deep_patch_redis() → None

Patch Redis at a deeper level to guarantee all Redis operations are tracked. This patches the Redis module’s core methods that all clients use.