Utils Package

# Submodules section with toctree removed as automodule below will cover them. # ———- # .. toctree:: # :maxdepth: 1 # # utils.cache # utils.helpers # utils.logging # utils.period_parser # utils.ratelimiter # utils.retry_decorators # utils.sentry_integration

class utils.cache.SheetCache

Базовые классы: object

A simple in-memory cache with Time-To-Live (TTL) for sheet data.

__init__()

Initializes the cache store.

utils.cache.cached_sheet_method(key_fn)

Decorator to cache the result of a method reading from Sheets.

The decorated method must have user_id as its first or second argument (after self). The key_fn is a lambda that receives the method’s arguments (excluding self and user_id) and should return a string that, combined with user_id and the method name, forms a unique cache key.

Параметры:

key_fn (Callable[..., str]) – A callable that generates a unique key suffix from method args.

Результат:

A decorator function.

utils.cache.invalidate_sheet_cache(user_id)

Invalidates all cache entries for a specific user.

This should be called after any operation that modifies a user’s spreadsheet to ensure cache consistency.

Параметры:

user_id (int) – The Telegram ID of the user whose cache entries should be cleared.

Тип результата:

None

General helper functions used across the application.

utils.helpers.escape_markdown_v2(text)

Escapes special characters for MarkdownV2.

See: https://core.telegram.org/bots/api#markdownv2-style

Параметры:

text (str)

Тип результата:

str

utils.helpers.format_date(date_obj, tz=None)

Formats a datetime object to DD.MM.YYYY string in the specified timezone.

Параметры:

• date_obj (datetime)

• tz (str | None)

Тип результата:

str

utils.helpers.get_day_of_week(date_obj, tz=None)

Gets the localized (Russian) name of the day of the week for a datetime object.

Параметры:

• date_obj (datetime)

• tz (str | None)

Тип результата:

str

Logging setup for the application using structlog and standard logging.

utils.logging.setup_logging(log_level='INFO')

Sets up standard logging and structlog with JSON output.

Параметры:

log_level (str | int) – The logging level (str or int).

Результат:

The root structlog logger to which context can be bound.

Тип результата:

BoundLogger

utils.period_parser.parse_period(text)

Returns the number of days. Raises ValueError on failure.

Параметры:

text (str)

Тип результата:

int

Rate limiting utilities using the Token Bucket algorithm.

This module provides:

• RateLimitException: Custom exception raised when a rate limit is exceeded.

• TokenBucket: Core implementation of the token bucket algorithm, managing token replenishment and consumption for a single resource.

• UserRateLimiter: Manages multiple TokenBucket instances, typically one per user, to enforce user-specific rate limits for operations like LLM calls.

The UserRateLimiter is intended to be instantiated once per limited resource type (e.g., one for LLM calls) and then used by calling its check_limit method, passing the user identifier.

exception utils.ratelimiter.RateLimitException(message, retry_after_seconds=None)

Базовые классы: Exception

Exception raised when an operation exceeds the defined rate limit.

Параметры:

• message (str)

• retry_after_seconds (float | None)

class utils.ratelimiter.TokenBucket(tokens_per_second, max_tokens)

Базовые классы: object

Implements the Token Bucket algorithm for a single rate-limited resource.

tokens_per_second

The rate at which tokens are added to the bucket.

max_tokens

The maximum capacity of the bucket.

current_tokens

The current number of available tokens.

last_update_time

The timestamp of the last token refill.

Параметры:

• tokens_per_second (float)

• max_tokens (float)

__init__(tokens_per_second, max_tokens)

Initializes a TokenBucket.

Параметры:

• tokens_per_second (float) – Rate of token replenishment.

• max_tokens (float) – Maximum token capacity.

consume(tokens_to_consume=1)

Attempts to consume a number of tokens.

Параметры:

tokens_to_consume (int) – The number of tokens to attempt to consume (must be > 0).

Тип результата:

bool

Результат:

True if tokens were consumed successfully, False otherwise.

get_retry_after(tokens_to_consume=1)

Calculates how long to wait before tokens_to_consume can be met.

Параметры:

tokens_to_consume (int) – The number of tokens desired.

Тип результата:

float | None

Результат:

Time in seconds to wait, or 0 if consumable immediately, or None if the request exceeds max_tokens.

class utils.ratelimiter.UserRateLimiter(default_tokens_per_second, default_max_tokens)

Базовые классы: object

Manages a collection of TokenBuckets, typically one per user ID.

This allows enforcing rate limits on a per-user basis for shared resources.

Параметры:

• default_tokens_per_second (float)

• default_max_tokens (float)

__init__(default_tokens_per_second, default_max_tokens)

Initializes the UserRateLimiter with default bucket parameters.

Параметры:

• default_tokens_per_second (float) – Default token refill rate for new user buckets.

• default_max_tokens (float) – Default maximum token capacity for new user buckets.

check_limit(user_id, tokens_to_consume=1)

Checks if the specified user can consume the given number of tokens.

Параметры:

• user_id (Any) – The identifier of the user (e.g., Telegram user ID).

• tokens_to_consume (int) – The number of tokens to attempt to consume.

Исключение:

RateLimitException – If the user’s bucket does not have enough tokens.

Тип результата:

None

Standardized Tenacity retry decorators for external API calls.

This module provides pre-configured tenacity.retry decorators for common API interaction patterns within the application, such as calls to Google Sheets and OpenAI LLM. It includes a shared logging callback (_log_retry) to report retry attempts.

Key decorators: - retry_google_sheets: For gspread API calls. - retry_openai_llm: For OpenAI API calls, reraises exceptions. - retry_openai_llm_no_reraise: For OpenAI calls where returning None on failure is preferred.

Sentry integration setup for error tracking and performance monitoring.

utils.sentry_integration.setup_sentry()

Initializes Sentry if the SENTRY_DSN environment variable is set.

Тип результата:

None