Handler - Referencia de API

Este módulo proporciona handlers para logging y notificaciones.

LoggingHandler

class ctrutils.handler.LoggingHandler(level=20, message_format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', logger_name=None)[fuente]

Bases: object

Clase para configurar y manejar logs con soporte para múltiples handlers.

Esta clase permite crear y personalizar distintos handlers para el registro de mensajes, incluyendo salida a consola, almacenamiento en archivos (con rotación por tamaño o por tiempo), envío a Grafana Loki y notificaciones vía Telegram.

Cada instancia crea un logger único (mediante un nombre generado automáticamente o especificado) y desactiva la propagación de los mensajes al logger raíz, evitando duplicación.

Parámetros:

level (int) – Nivel del logger (por defecto, logging.INFO).
message_format (str) – Formato de los mensajes de log.
logger_name (Optional[str]) – Nombre del logger. Si no se especifica, se genera uno único.

Ejemplo:

handler = LoggingHandler(level=logging.DEBUG)
console_handler = handler.create_stream_handler()
logger = handler.add_handlers([console_handler])
logger.debug("Mensaje de debug")

__init__(level=20, message_format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', logger_name=None)[fuente]

Inicializa el LoggingHandler.

Parámetros:

level (int) – Nivel de logging (DEBUG, INFO, WARNING, ERROR, CRITICAL).
message_format (str) – Formato de los mensajes.
logger_name (str | None) – Nombre único del logger.

create_stream_handler()[fuente]

Crea un handler para la salida de logs a la consola.

Devuelve:: Instancia configurada de StreamHandler.
Tipo del valor devuelto:: StreamHandler

Ejemplo:

console_handler = handler.create_stream_handler()

create_file_handler(log_file)[fuente]

Crea un handler para guardar los logs en un archivo plano.

Parámetros:: log_file (str) – Ruta del archivo donde se almacenarán los logs.
Devuelve:: Instancia configurada de FileHandler.
Tipo del valor devuelto:: FileHandler

Ejemplo:

file_handler = handler.create_file_handler("app.log")

create_size_rotating_file_handler(log_file, max_bytes, backup_count)[fuente]

Crea un handler para guardar los logs en un archivo con rotación basada en tamaño.

La rotación por tamaño es útil para evitar que los archivos de log crezcan demasiado y para mantener un número limitado de respaldos.

Parámetros:

log_file (str) – Ruta del archivo de log.
max_bytes (int) – Tamaño máximo en bytes antes de que se produzca la rotación.
backup_count (int) – Número máximo de archivos de respaldo a mantener.

Devuelve:

Instancia configurada de RotatingFileHandler.

Tipo del valor devuelto:

RotatingFileHandler

Ejemplo:

rotating_handler = handler.create_size_rotating_file_handler(
    log_file="rotating.log",
    max_bytes=10*1024*1024,  # 10MB
    backup_count=5
)

create_timed_rotating_file_handler(log_file, when, interval, backup_count)[fuente]

Crea un handler para guardar los logs en un archivo con rotación basada en tiempo.

La rotación por tiempo es útil para generar archivos de log separados por períodos definidos (por ejemplo, diarios o semanales).

Parámetros:

log_file (str) – Ruta del archivo de log.
when (str) – Unidad de tiempo para la rotación (“S”=segundos, “M”=minutos, “H”=horas, “D”=días).
interval (int) – Intervalo de tiempo para la rotación.
backup_count (int) – Número máximo de archivos de respaldo a mantener.

Devuelve:

Instancia configurada de TimedRotatingFileHandler.

Tipo del valor devuelto:

TimedRotatingFileHandler

Ejemplo:

timed_handler = handler.create_timed_rotating_file_handler(
    log_file="timed.log",
    when="D",  # Diario
    interval=1,
    backup_count=7
)

create_loki_handler(url, labels=None, level=20, timeout=5)[fuente]

Crea un handler para enviar logs a Grafana Loki.

Loki es un sistema de agregación de logs horizontalmente escalable, altamente disponible y multi-tenant inspirado en Prometheus.

Parámetros:

url (str) – URL del servidor Loki (ej: «http://localhost:3100»).
labels (Optional[dict]) – Diccionario de labels para identificar los logs en Loki.
level (int) – Nivel mínimo de log para enviar a Loki.
timeout (int) – Timeout para las peticiones HTTP en segundos.

Devuelve:

Instancia configurada de LokiHandler.

Tipo del valor devuelto:

LokiHandler

Muestra:

ImportError – Si no está disponible el módulo loki_handler.

Ejemplo:

loki_handler = handler.create_loki_handler(
    url="http://loki:3100",
    labels={"app": "myapp", "env": "production"},
    level=logging.INFO
)

create_telegram_handler(token, chat_id, level=40, parse_mode='HTML', timeout=20)[fuente]

Crea un handler para enviar logs a un chat o canal de Telegram.

Importante:

Especifica el nivel mínimo de log (parámetro level) para controlar cuándo se envían las alertas.
Por ejemplo, si se establece a logging.CRITICAL, solo se enviarán notificaciones para mensajes críticos.
Para recibir notificaciones de diferentes niveles de forma independiente, crea otra instancia de LoggingHandler o agrega este handler a otro logger.

Parámetros:

token (str) – Token de autenticación del bot de Telegram.
chat_id (str) – ID del chat o canal de Telegram donde se enviarán los mensajes.
level (int) – Nivel mínimo de log para enviar mensajes vía Telegram (por defecto, logging.ERROR).
parse_mode (str) – Modo de parseo para el formato del mensaje («HTML», «Markdown», «MarkdownV2»).
timeout (int) – Timeout para las peticiones HTTP en segundos.

Devuelve:

Instancia configurada de TelegramBotHandler.

Tipo del valor devuelto:

TelegramBotHandler

Muestra:

ImportError – Si no está disponible el módulo telegram_handler.

Ejemplo:

telegram_handler = handler.create_telegram_handler(
    token="YOUR_TELEGRAM_BOT_TOKEN",
    chat_id="YOUR_TELEGRAM_CHAT_ID",
    level=logging.ERROR
)

add_handlers(handlers, logger_name=None)[fuente]

Agrega los handlers proporcionados a un logger y lo devuelve.

Este método asocia los handlers a un logger identificado por el nombre de la instancia o uno proporcionado.

Parámetros:

handlers (List[logging.Handler]) – Lista de instancias de logging.Handler a asociar.
logger_name (Optional[str]) – Nombre del logger. Si no se proporciona, se utiliza el nombre único de la instancia.

Devuelve:

El logger configurado con los handlers asociados.

Tipo del valor devuelto:

logging.Logger

Ejemplo:

logger = handler.add_handlers([console_handler, file_handler])

remove_handlers(remove_all=True, handler_types=None)[fuente]

Elimina uno o varios handlers asociados al logger de la instancia.

Parámetros:

remove_all (bool) – Si es True, elimina todos los handlers asociados al logger.
handler_types (Optional[List[type]]) – Lista de clases de handlers a eliminar.

Muestra:

ValueError – Si no se ha configurado ningún logger previamente.

Ejemplo:

# Eliminar todos los handlers
handler.remove_handlers(remove_all=True)

# Eliminar solo los handlers de tipo StreamHandler
handler.remove_handlers(remove_all=False, handler_types=[StreamHandler])

static quick_console_logger(name='app', level=20)[fuente]

Crea rápidamente un logger con salida a consola.

Parámetros:

name (str) – Nombre del logger.
level (int) – Nivel de logging.

Devuelve:

Logger configurado.

Tipo del valor devuelto:

logging.Logger

Ejemplo:

logger = LoggingHandler.quick_console_logger("myapp", logging.DEBUG)
logger.debug("Debug message")

static quick_file_logger(name='app', log_file='app.log', level=20)[fuente]

Crea rápidamente un logger con salida a consola y archivo.

Parámetros:

name (str) – Nombre del logger.
log_file (str) – Ruta del archivo de log.
level (int) – Nivel de logging.

Devuelve:

Logger configurado.

Tipo del valor devuelto:

logging.Logger

Ejemplo:

logger = LoggingHandler.quick_file_logger("myapp", "app.log")
logger.info("Info message")

static production_logger(name, log_file, loki_url=None, loki_labels=None, telegram_token=None, telegram_chat_id=None, level=20)[fuente]

Crea un logger completo para producción con múltiples outputs.

Incluye: - Archivo con rotación por tamaño (10MB, 5 backups) - Loki (opcional, si se proporcionan credenciales) - Telegram (opcional, si se proporcionan credenciales, solo errores)

Parámetros:

name (str) – Nombre del logger.
log_file (str) – Ruta del archivo de log.
loki_url (Optional[str]) – URL del servidor Loki (opcional).
loki_labels (Optional[dict]) – Labels para Loki (opcional).
telegram_token (Optional[str]) – Token del bot de Telegram (opcional).
telegram_chat_id (Optional[str]) – Chat ID de Telegram (opcional).
level (int) – Nivel de logging.

Devuelve:

Logger configurado para producción.

Tipo del valor devuelto:

logging.Logger

Ejemplo:

logger = LoggingHandler.production_logger(
    name="myapp",
    log_file="production.log",
    loki_url="http://loki:3100",
    loki_labels={"app": "myapp", "env": "prod"},
    telegram_token="YOUR_TOKEN",
    telegram_chat_id="YOUR_CHAT_ID"
)

log_exception_and_exit(exception, exit_code=1, context=None)[fuente]

Logea una excepción con contexto y termina el proceso.

Útil para tareas de scheduler que necesitan exit codes específicos.

Parámetros:

exception (Exception) – Excepción a loguear.
exit_code (int) – Código de salida del proceso.
context (Optional[dict]) – Contexto adicional a incluir en el log.

Ejemplo:

try:
    process_data()
except Exception as e:
    handler.log_exception_and_exit(
        e,
        exit_code=1,
        context={"task": "data_processing"}
    )

LokiHandler

class ctrutils.handler.LokiHandler(url, labels, level=20, timeout=5, batch_size=0)[fuente]

Bases: Handler

Handler personalizado para enviar logs a Grafana Loki.

Este handler permite enviar logs a un servidor Loki utilizando su API HTTP. Los logs se etiquetan con labels (similares a Prometheus) para facilitar consultas y filtrado en Grafana.

Características:

Envío en tiempo real o con batching
Labels personalizados por aplicación/servicio/entorno
Manejo de errores sin bloquear la aplicación
Timeout configurable

Niveles de log recomendados:

logging.DEBUG (10): Depuración detallada
logging.INFO (20): Información general
logging.WARNING (30): Advertencias
logging.ERROR (40): Errores
logging.CRITICAL (50): Errores críticos

Parámetros:

url (str) – URL del servidor Loki (ej: «http://localhost:3100»).
labels (Dict[str, str]) – Diccionario de labels para identificar los logs. Ejemplo: {«app»: «myapp», «env»: «production», «host»: «server1»}
level (int, optional) – Nivel mínimo de log para enviar a Loki. Defaults to logging.INFO.
timeout (int, optional) – Tiempo de espera para la solicitud HTTP en segundos. Defaults to 5.
batch_size (int, optional) – Número de logs a acumular antes de enviar (0 = sin batching). Defaults to 0.

Variables:

url – URL del endpoint de push de Loki.
labels – Labels asociados a este handler.
timeout – Timeout configurado para las peticiones HTTP.
batch_size – Tamaño del batch configurado.
batch – Lista temporal para acumular logs (si batching está habilitado).

Ejemplo básico:

import logging
from loki_handler import LokiHandler

handler = LokiHandler(
    url="http://localhost:3100",
    labels={"app": "myapp", "env": "dev"},
    level=logging.INFO
)

logger = logging.getLogger("myapp")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info("Aplicación iniciada")
logger.error("Error al procesar datos")

Ejemplo con batching:

handler = LokiHandler(
    url="http://loki:3100",
    labels={"app": "myapp", "env": "production"},
    level=logging.INFO,
    batch_size=10  # Envía cada 10 logs
)

Ejemplo en producción:

from ctrutils.handler import LoggingHandler

logger = LoggingHandler.production_logger(
    name="myapp",
    log_file="app.log",
    loki_url="http://loki:3100",
    loki_labels={
        "app": "myapp",
        "env": "production",
        "host": "server-01",
        "version": "1.0.0"
    }
)

logger.info("Aplicación en producción iniciada")

__init__(url, labels, level=20, timeout=5, batch_size=0)[fuente]

Inicializa el LokiHandler.

Parámetros:

url (str) – URL del servidor Loki (sin el path /loki/api/v1/push).
labels (Dict[str, str]) – Diccionario de labels para los logs.
level (int) – Nivel mínimo de log.
timeout (int) – Timeout en segundos para las peticiones HTTP.
batch_size (int) – Número de logs a acumular antes de enviar (0 = sin batching).

emit(record)[fuente]

Envía el mensaje de log a Loki.

Este método es llamado automáticamente por el logger cuando se registra un mensaje que cumple con el nivel mínimo configurado.

Si el batching está habilitado, acumula los logs hasta alcanzar batch_size antes de enviarlos. Si no, envía inmediatamente.

Parámetros:: record (logging.LogRecord) – Registro de log a enviar.

flush()[fuente]

Fuerza el envío de todos los logs pendientes en el batch.

Este método debe ser llamado antes de cerrar la aplicación si se usa batching, para asegurar que todos los logs se envíen.

Ejemplo:

import atexit

loki_handler = LokiHandler(...)
logger.addHandler(loki_handler)

# Asegurar que se envíen los logs pendientes al cerrar
atexit.register(loki_handler.flush)

close()[fuente]

Cierra el handler enviando los logs pendientes.

TelegramBotHandler

class ctrutils.handler.TelegramBotHandler(token, chat_id, level=40, parse_mode='HTML', timeout=20)[fuente]

Bases: Handler

Handler personalizado para enviar logs a un chat de Telegram.

Este handler permite enviar mensajes de log a un chat o canal de Telegram utilizando un bot. El nivel de log y el modo de parse pueden ser configurados. Se recomienda utilizar los niveles predefinidos de logging para mayor claridad.

Niveles de log disponibles:

logging.DEBUG (10): Mensajes de depuración detallados
logging.INFO (20): Mensajes informativos generales
logging.WARNING (30): Advertencias que no detienen el programa
logging.ERROR (40): Errores que afectan la funcionalidad
logging.CRITICAL (50): Errores críticos que pueden detener el programa

Parse Modes:

HTML: Permite usar HTML básico (<b>, <i>, <code>, <pre>)
Markdown: Markdown básico (legacy)
MarkdownV2: Markdown mejorado con más opciones

Importante:

Configura el nivel apropiado para evitar spam en Telegram
Por defecto envía solo ERROR y superiores
Para testing usa un chat privado, no grupos

Parámetros:

token (str) – Token de autenticación del bot de Telegram.
chat_id (str) – ID del chat o canal de Telegram donde se enviarán los mensajes.
level (int, optional) – Nivel mínimo de log para enviar mensajes. Defaults to logging.ERROR.
parse_mode (ParseMode, optional) – Modo de parse para el formato del mensaje. Defaults to HTML.
timeout (int, optional) – Tiempo de espera para la solicitud HTTP en segundos. Defaults to 20.

Variables:

token – Token de autenticación del bot de Telegram.
chat_id – ID del chat o canal de Telegram.
parse_mode – Modo de parse configurado.
timeout – Timeout configurado para las peticiones HTTP.

Ejemplo básico:

import logging
from telegram_handler import TelegramBotHandler

handler = TelegramBotHandler(
    token="123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11",
    chat_id="123456789",
    level=logging.ERROR,
    parse_mode="HTML"
)

logger = logging.getLogger("my_logger")
logger.addHandler(handler)
logger.setLevel(logging.ERROR)

logger.error("Este mensaje será enviado a Telegram")

Ejemplo con formato HTML:

handler = TelegramBotHandler(
    token="YOUR_TOKEN",
    chat_id="YOUR_CHAT_ID",
    parse_mode="HTML"
)

logger = logging.getLogger("app")
logger.addHandler(handler)

# Los mensajes pueden incluir HTML básico
logger.error("<b>Error crítico:</b> <code>División por cero</code>")

Ejemplo en producción:

from ctrutils.handler import LoggingHandler

logger = LoggingHandler.production_logger(
    name="myapp",
    log_file="production.log",
    telegram_token="YOUR_TOKEN",
    telegram_chat_id="YOUR_CHAT_ID"
)

# Solo errores y críticos se envían a Telegram
logger.info("Info - NO se envía a Telegram")
logger.error("Error - SÍ se envía a Telegram")

__init__(token, chat_id, level=40, parse_mode='HTML', timeout=20)[fuente]

Inicializa el TelegramBotHandler.

Parámetros:

token (str) – Token de autenticación del bot de Telegram.
chat_id (str) – ID del chat o canal de Telegram.
level (int) – Nivel mínimo de log para enviar mensajes.
parse_mode (Literal['HTML', 'Markdown', 'MarkdownV2']) – Modo de parse del mensaje («HTML», «Markdown», «MarkdownV2»).
timeout (int) – Timeout en segundos para las peticiones HTTP.

emit(record)[fuente]

Envía el mensaje de log a Telegram.

Este método es llamado automáticamente por el logger cuando se registra un mensaje que cumple con el nivel mínimo configurado.

El mensaje incluye: - Emoji según el nivel (ℹ️ INFO, ⚠️ WARNING, ❌ ERROR, 🚨 CRITICAL) - Timestamp - Nombre del logger - Nivel - Mensaje formateado

Parámetros:: record (logging.LogRecord) – Registro de log a enviar.