Skip to main content

Logs y monitorización de recursos

Project description

TAI-Alphi

Librería de logs y monitorización de procesos

¿Para qué sirve?

TAI-Alphi es una librería de logging avanzada que te permite:

  • 📝 Escribir logs una vez y enviarlos a múltiples destinos
  • ⚙️ Configurar múltiples loggers con diferentes comportamientos
  • 🔄 Reconfigurar loggers dinámicamente sin reiniciar la aplicación
  • 🎯 Personalizar niveles, formatos y destinos por logger
  • 🏗️ Patrón Singleton para gestión centralizada

Instalación

Poetry

poetry add tai-alphi

Pip

pip install tai-alphi

Dependencias

Requeridas (se instalan automáticamente):

  • pydantic ^2.11

Opcionales (instalar según necesidad):

  • pymongo - Para enviar logs a MongoDB/CosmosDB
  • logtail - Para enviar logs a BetterStack

Uso Básico

Inicio Rápido

from tai_alphi import Alphi

# Crear instancia (Singleton)
bot = Alphi()

# Obtener logger
logger = bot.get_logger()

# Usar logger
logger.debug('Mensaje de depuración')
logger.info('Mensaje informativo')
logger.warning('Mensaje de advertencia')
logger.error('Mensaje de error')
logger.critical('Mensaje crítico')

Con Configuración Inicial

from tai_alphi import Alphi

# Configurar al inicio
config = {
    'app': {
        'consola': {
            'enabled': True,
            'log_level': 'INFO',
            'display_info': ['asctime', 'levelname'],
            'time_format': '%H:%M:%S'
        }
    }
}

# Crear instancia con configuración
alphi = Alphi(config)

# Obtener logger configurado
logger = alphi.get_logger('app')

logger.info('Aplicación iniciada')

Destinos (Handlers)

TAI-Alphi puede enrutar logs hacia 4 destinos diferentes:

Destino Descripción Configuración Requerida
🖥️ Consola Salida estándar (terminal) Ninguna (por defecto)
💬 Teams Microsoft Teams Webhook URL
🗄️ NoSQL MongoDB / CosmosDB Credenciales de conexión
📊 LogTail BetterStack Token de acceso

Cada destino se puede configurar individualmente con diferentes niveles de log, formatos y filtros.

Configuración

Estructura de Configuración

La configuración se define como un diccionario Python con la siguiente estructura:

config = {
    'nombre_logger': {
        'consola': {
            'enabled': bool,           # Activar/desactivar
            'log_level': str,          # DEBUG | INFO | WARN | ERROR | CRIT
            'display_info': list[str], # Campos a mostrar
            'time_format': str         # Formato de timestamp
        },
        'teams': {
            'enabled': bool,
            'log_level': str,
            'display_info': list[str],
            'time_format': str,
            'project': str,            # Nombre del proyecto
            'pipeline': str,           # Nombre del pipeline
            'notifications': list[str] # Emails a notificar
        },
        'nosql': {
            'enabled': bool,
            'log_level': str,
            'display_info': list[str],
            'time_format': str,
            'expiration': float        # Días hasta expiración (opcional)
        },
        'logtail': {
            'enabled': bool,
            'log_level': str
        }
    }
}

Parámetros de Configuración

Niveles de Log (log_level)

  • 'DEBUG' - Información detallada para diagnóstico
  • 'INFO' - Eventos informativos generales
  • 'WARN' - Advertencias, requieren atención
  • 'ERROR' - Errores que afectan funcionalidad
  • 'CRIT' - Errores críticos que requieren acción inmediata

Información a Mostrar (display_info)

Opciones disponibles:

  • 'asctime' - Timestamp del log
  • 'filename' - Nombre del archivo fuente
  • 'funcName' - Nombre de la función
  • 'levelname' - Nivel del log
  • 'lineno' - Número de línea
  • 'module' - Nombre del módulo
  • 'pathname' - Ruta completa del archivo

Formato de Tiempo (time_format)

Usa formato estándar de Python strftime. Ejemplos:

  • '%H:%M:%S' - 14:30:45
  • '%Y-%m-%d %H:%M:%S' - 2025-11-18 14:30:45
  • '%d/%m/%Y %H:%M' - 18/11/2025 14:30

Documentación completa de strftime

Configuración de Credenciales

Para usar Teams, NoSQL o LogTail, configura las credenciales:

from tai_alphi import Alphi
import os

# Configurar al inicio de la aplicación
Alphi(config)

# Configurar Teams (método de clase)
Alphi.set_teams(webhook=os.getenv('TEAMS_WEBHOOK'))

# Configurar MongoDB/CosmosDB (método de clase)
Alphi.set_nosql(
    user=os.getenv('DB_USER'),
    pwd=os.getenv('DB_PASSWORD'),
    host=os.getenv('DB_HOST', 'localhost'),
    port=int(os.getenv('DB_PORT', '27017')),
    db_name='logs_db',
    collection_name='application_logs'
)

# Configurar LogTail (método de clase)
Alphi.set_logtail(token=os.getenv('LOGTAIL_TOKEN'))

[!TIP] Usa variables de entorno para las credenciales por seguridad

Uso Avanzado

Múltiples Loggers

Puedes configurar y usar múltiples loggers con diferentes comportamientos:

from tai_alphi import Alphi

config = {
    'app': {
        'consola': {
            'enabled': True,
            'log_level': 'INFO',
            'display_info': ['asctime', 'levelname']
        }
    },
    'database': {
        'consola': {
            'enabled': True,
            'log_level': 'DEBUG',
            'display_info': ['asctime', 'levelname', 'module']
        },
        'nosql': {
            'enabled': True,
            'log_level': 'WARN'
        }
    },
    'api': {
        'consola': {
            'enabled': True,
            'log_level': 'WARN'
        },
        'teams': {
            'enabled': True,
            'log_level': 'ERROR',
            'project': 'Mi API',
            'pipeline': 'Producción'
        }
    }
}

alphi = Alphi(config)

# Configurar credenciales (métodos de clase)
Alphi.set_nosql(user='...', pwd='...', host='...', port=27017, 
                db_name='logs', collection_name='logs')
Alphi.set_teams(webhook='https://...')

# Obtener y usar cada logger
app_logger = alphi.get_logger('app')
db_logger = alphi.get_logger('database')
api_logger = alphi.get_logger('api')

app_logger.info('Aplicación iniciada')        # Solo consola
db_logger.debug('Query ejecutado')            # Solo consola
db_logger.warning('Pool al 90%')              # Consola + NoSQL
api_logger.error('Timeout en API externa')    # Consola + Teams

Acceso Directo con Método de Clase

from tai_alphi import Alphi

# Configurar una vez
Alphi(config)

# Usar en cualquier módulo sin instanciar
logger = Alphi.get_logger_by_name('app')
logger.info('Acceso directo al logger')

Configuración Dinámica

Configurar Antes de Crear

from tai_alphi import Alphi

alphi = Alphi()

# Configurar logger antes de usarlo
Alphi.configure_logger('email-service', {
    'consola': {
        'enabled': True,
        'log_level': 'DEBUG',
        'display_info': ['asctime', 'levelname', 'funcName']
    }
})

# Crear logger (usa la configuración preestablecida)
email_logger = alphi.get_logger('email-service')
email_logger.debug('Conectando a servidor SMTP')

Reconfigurar Logger Existente

from tai_alphi import Alphi

alphi = Alphi()

# Crear logger con configuración por defecto
logger = alphi.get_logger('notifications')
logger.info('Notificación enviada')

# Reconfigurar en caliente
Alphi.configure_logger('notifications', {
    'consola': {
        'enabled': True,
        'log_level': 'DEBUG',
        'display_info': ['asctime', 'levelname', 'module', 'funcName']
    }
})

# El mismo logger ahora usa la nueva configuración
logger.debug('Este mensaje ahora aparece!')

[!IMPORTANT] La reconfiguración es automática. No necesitas obtener el logger nuevamente.

Configurar Múltiples Loggers Simultáneamente

from tai_alphi import Alphi

configs = {
    'auth': {
        'consola': {'enabled': True, 'log_level': 'INFO'}
    },
    'payment': {
        'consola': {'enabled': True, 'log_level': 'WARN'}
    },
    'analytics': {
        'consola': {'enabled': True, 'log_level': 'ERROR'}
    }
}

Alphi.configure_multiple_loggers(configs)

# Usar los loggers
auth = Alphi.get_logger_by_name('auth')
payment = Alphi.get_logger_by_name('payment')
analytics = Alphi.get_logger_by_name('analytics')

Sobreescritura de Configuración Global

Puedes reemplazar completamente la configuración de Alphi:

from tai_alphi import Alphi

# Configuración inicial
config1 = {
    'service-a': {'consola': {'enabled': True, 'log_level': 'INFO'}},
    'service-b': {'consola': {'enabled': True, 'log_level': 'DEBUG'}}
}

alphi = Alphi(config1)
logger = alphi.get_logger('service-a')

# Sobreescribir con nueva configuración
config2 = {
    'service-x': {'consola': {'enabled': True, 'log_level': 'WARN'}},
    'service-y': {'consola': {'enabled': True, 'log_level': 'ERROR'}}
}

alphi = Alphi(config2)  # Misma instancia Singleton, nueva configuración

# Los loggers anteriores ya no existen en la configuración
# Los nuevos loggers usan la configuración actualizada
logger_x = alphi.get_logger('service-x')

Cambiar Colección NoSQL por Logger

from tai_alphi import Alphi

Alphi(config)
Alphi.set_nosql(user='...', pwd='...', host='...', port=27017,
                db_name='logs', collection_name='default_logs')

# Logger con colección por defecto
logger1 = Alphi.get_logger_by_name('service1')
logger1.info('Va a default_logs')

# Logger con colección específica
logger2 = Alphi.get_logger_by_name('service2')
logger2.set_nosql_collection(collection_name='service2_logs')
logger2.info('Va a service2_logs')

Cambiar Webhook de Teams por Logger

from tai_alphi import Alphi

Alphi(config)

# Logger con webhook personalizado
logger = Alphi.get_logger_by_name('critical-service')
logger.set_teams_webhook(webhook='https://teams.webhook.especifico')
logger.error('Este error va a un canal específico de Teams')

Inspección y Gestión

Listar Loggers Configurados

from tai_alphi import Alphi

alphi = Alphi.get_instance()

# Ver todos los loggers y su origen
configured = alphi.list_configured_loggers()
for name, source in configured.items():
    print(f"{name}: {source}")
# Salida:
# app: config
# database: dynamic
# api: config

# Ver nombres disponibles
print(alphi.get_available_logger_names())
# ['tai-logger', 'app', 'database', 'api']

# Verificar si existe configuración específica
print(alphi.has_logger_config('app'))  # True
print(alphi.has_logger_config('inexistente'))  # False

Obtener Configuración de un Logger

from tai_alphi import Alphi

alphi = Alphi.get_instance()

# Obtener configuración completa
config = alphi.get_logger_config('app')

print(f"Consola habilitada: {config.consola.enabled}")
print(f"Nivel de log: {config.consola.log_level}")
print(f"Display info: {config.consola.display_info}")
print(f"Time format: {config.consola.time_format}")

Remover Configuración Dinámica

from tai_alphi import Alphi

alphi = Alphi.get_instance()

# Configurar dinámicamente
Alphi.configure_logger('temp', {
    'consola': {'enabled': True, 'log_level': 'DEBUG'}
})

logger = alphi.get_logger('temp')
logger.debug('Mensaje con DEBUG')

# Remover configuración dinámica (vuelve a la por defecto)
alphi.remove_logger_config('temp')

logger.info('Ahora usa configuración por defecto')

Ver Loggers en Caché

from tai_alphi import Alphi

alphi = Alphi.get_instance()

# Ver todos los loggers creados
for name, logger in alphi.loggers.items():
    print(f"{name}: {logger}")

Patrón Singleton

Alphi implementa el patrón Singleton, garantizando una única instancia en toda la aplicación:

from tai_alphi import Alphi

alphi1 = Alphi()
alphi2 = Alphi()
alphi3 = Alphi.get_instance()

# Todas son la misma instancia
assert alphi1 is alphi2 is alphi3  # True

# Para tests o reinicios, puedes resetear
Alphi.reset_instance()

Valores por Defecto

Si no se especifica configuración, cada handler usa estos valores:

Consola

{
    'enabled': True,
    'log_level': 'INFO',
    'display_info': ['asctime', 'levelname'],
    'time_format': '%H:%M:%S'
}

Teams

{
    'enabled': False,
    'log_level': 'INFO',
    'display_info': ['asctime', 'levelname'],
    'time_format': '%H:%M:%S',
    'project': 'Project Name',
    'pipeline': 'Pipeline Name',
    'notifications': []
}

NoSQL

{
    'enabled': False,
    'log_level': 'INFO',
    'display_info': ['asctime', 'levelname'],
    'time_format': '%Y-%m-%d %H:%M:%S',
    'expiration': None  # Sin expiración
}

LogTail

{
    'enabled': False,
    'log_level': 'INFO'
}

Mejores Prácticas

  1. Un logger por componente: Crea loggers específicos para cada módulo importante de tu aplicación

    app_logger = Alphi.get_logger_by_name('app')
    db_logger = Alphi.get_logger_by_name('database')
    api_logger = Alphi.get_logger_by_name('api')
    
  2. Niveles apropiados: Usa DEBUG en desarrollo, INFO o superior en producción

    import os
    
    log_level = 'DEBUG' if os.getenv('ENV') == 'dev' else 'INFO'
    config = {'app': {'consola': {'log_level': log_level}}}
    
  3. Variables de entorno: Almacena credenciales de forma segura

    Alphi.set_teams(webhook=os.getenv('TEAMS_WEBHOOK'))
    Alphi.set_nosql(
        user=os.getenv('DB_USER'),
        pwd=os.getenv('DB_PASSWORD'),
        # ...
    )
    
  4. Configuración dinámica: Ajusta comportamiento sin reiniciar

    # Activar debug temporalmente para investigar
    Alphi.configure_logger('app', {
        'consola': {'log_level': 'DEBUG'}
    })
    
  5. Acceso directo: Usa get_logger_by_name() para acceso rápido

    # En cualquier módulo
    logger = Alphi.get_logger_by_name('app')
    logger.info('Mensaje')
    

Ejemplos Completos

Aplicación Web con Múltiples Componentes

from tai_alphi import Alphi
import os

# Configuración al inicio
config = {
    'web': {
        'consola': {'enabled': True, 'log_level': 'INFO'},
        'teams': {'enabled': True, 'log_level': 'ERROR', 
                  'project': 'Web App', 'pipeline': 'Production'}
    },
    'db': {
        'consola': {'enabled': True, 'log_level': 'WARN'},
        'nosql': {'enabled': True, 'log_level': 'ERROR', 'expiration': 90.0}
    },
    'api': {
        'consola': {'enabled': True, 'log_level': 'INFO'},
        'teams': {'enabled': True, 'log_level': 'CRIT',
                  'notifications': ['admin@example.com']}
    }
}

# Inicializar
Alphi(config)
Alphi.set_teams(webhook=os.getenv('TEAMS_WEBHOOK'))
Alphi.set_nosql(
    user=os.getenv('DB_USER'),
    pwd=os.getenv('DB_PASSWORD'),
    host=os.getenv('DB_HOST'),
    port=int(os.getenv('DB_PORT', '27017')),
    db_name='app_logs',
    collection_name='production_logs'
)

# Usar en diferentes módulos
web_logger = Alphi.get_logger_by_name('web')
db_logger = Alphi.get_logger_by_name('db')
api_logger = Alphi.get_logger_by_name('api')

# Logs de diferentes componentes
web_logger.info('Servidor web iniciado en puerto 8000')
db_logger.warning('Pool de conexiones al 85%')
api_logger.critical('API externa no responde - sistema degradado')

Desarrollo vs Producción

from tai_alphi import Alphi
import os

ENV = os.getenv('ENV', 'dev')

if ENV == 'dev':
    # Desarrollo: Solo consola, nivel DEBUG
    config = {
        'app': {
            'consola': {
                'enabled': True,
                'log_level': 'DEBUG',
                'display_info': ['asctime', 'levelname', 'funcName', 'lineno']
            }
        }
    }
else:
    # Producción: Consola + Teams + NoSQL
    config = {
        'app': {
            'consola': {
                'enabled': True,
                'log_level': 'INFO',
                'display_info': ['asctime', 'levelname']
            },
            'teams': {
                'enabled': True,
                'log_level': 'ERROR',
                'project': 'Production App',
                'notifications': ['team@example.com']
            },
            'nosql': {
                'enabled': True,
                'log_level': 'WARN',
                'expiration': 30.0
            }
        }
    }

Alphi(config)

if ENV != 'dev':
    Alphi.set_teams(webhook=os.getenv('TEAMS_WEBHOOK'))
    Alphi.set_nosql(
        user=os.getenv('DB_USER'),
        pwd=os.getenv('DB_PASSWORD'),
        host=os.getenv('DB_HOST'),
        port=int(os.getenv('DB_PORT')),
        db_name='logs',
        collection_name='app_logs'
    )

logger = Alphi.get_logger_by_name('app')

API Reference

Clase Alphi

Constructor

Alphi(settings: dict | None = None)

Crea o retorna la instancia Singleton de Alphi.

Parámetros:

  • settings (dict | None): Configuración de loggers

Métodos de Instancia

get_logger(logger_name: str = None, dev: bool = False, exec_info: bool = False) -> LoggerFactory

Obtiene o crea un logger.

Parámetros:

  • logger_name (str): Nombre del logger. Si es None, usa el primero configurado o el default
  • dev (bool): Si es True, desactiva todos los handlers excepto consola
  • exec_info (bool): Si es True, añade traceback a logs ERROR y CRITICAL

Retorna: Instancia de LoggerFactory


Métodos de Clase para Configuración

set_teams(webhook: str) -> str

Método de clase para configurar el webhook de Microsoft Teams.

Parámetros:

  • webhook (str): URL del webhook de Teams

Retorna: El webhook configurado

Ejemplo:

Alphi.set_teams(webhook='https://outlook.office.com/webhook/...')

set_nosql(user: str, pwd: str, host: str, port: int, db_name: str, collection_name: str) -> CosmosDB

Método de clase para configurar la conexión a MongoDB/CosmosDB.

Parámetros:

  • user (str): Usuario de la base de datos
  • pwd (str): Contraseña
  • host (str): Host del servidor
  • port (int): Puerto del servidor
  • db_name (str): Nombre de la base de datos
  • collection_name (str): Nombre de la colección

Retorna: Instancia de CosmosDB

Ejemplo:

Alphi.set_nosql(
    user='admin',
    pwd='password',
    host='localhost',
    port=27017,
    db_name='logs_db',
    collection_name='app_logs'
)

set_logtail(token: str) -> str

Método de clase para configurar el token de LogTail.

Parámetros:

  • token (str): Token de acceso de LogTail

Retorna: El token configurado

Ejemplo:

Alphi.set_logtail(token='your_logtail_token_here')

has_logger_config(logger_name: str) -> bool

Verifica si existe configuración específica para un logger.


get_available_logger_names() -> list[str]

Retorna todos los nombres de loggers configurados.


get_logger_config(logger_name: str) -> LoggerConfig

Obtiene la configuración de un logger específico.


list_configured_loggers() -> dict[str, str]

Lista todos los loggers con su origen de configuración ('config' o 'dynamic').


add_logger_config(logger_name: str, config: dict) -> None

Añade o actualiza configuración dinámica de un logger.


add_multiple_logger_configs(configs: dict[str, dict]) -> None

Añade o actualiza múltiples configuraciones dinámicas.


remove_logger_config(logger_name: str) -> None

Elimina configuración dinámica de un logger (vuelve a usar la por defecto).

Métodos de Clase

get_instance(settings: dict | None = None) -> Alphi

Obtiene la instancia Singleton. Si no existe, la crea. Si se proporciona settings, sobreescribe la configuración.


get_logger_by_name(logger_name: str = None, dev: bool = False, exec_info: bool = False) -> LoggerFactory

Obtiene un logger sin necesidad de instanciar Alphi explícitamente.


configure_logger(logger_name: str, config: dict) -> None

Configura un logger dinámicamente (método de clase).


configure_multiple_loggers(configs: dict[str, dict]) -> None

Configura múltiples loggers dinámicamente (método de clase).


reset_instance() -> None

Resetea la instancia Singleton (útil para testing).

Clase LoggerFactory

Extiende logging.Logger con funcionalidad adicional.

Métodos Adicionales

reconfigure(new_config: LoggerConfig, nosql: CosmosDB = None, logtail_token: str = None, teams_webhook: str = None) -> None

Reconfigura el logger sin crear una nueva instancia.


set_nosql_collection(db_name: str = None, collection_name: str = None) -> None

Cambia la colección de NoSQL para este logger específico.


set_teams_webhook(webhook: str) -> None

Cambia el webhook de Teams para este logger específico.

Tests

Ejecuta el test completo para ver todas las funcionalidades en acción:

python tests/test_completo.py

El test demuestra:

  • Patrón Singleton
  • Configuración inicial y dinámica
  • Múltiples loggers y handlers
  • Reconfiguración en caliente
  • Inspección de configuraciones
  • Sobreescritura global
  • Y mucho más...

Licencia

MIT

Contribuciones

Las contribuciones son bienvenidas. Por favor, abre un issue o pull request en el repositorio oficial.

Soporte

Para reportar bugs o solicitar nuevas funcionalidades, visita: https://github.com/triplealpha-innovation/tai-alphi/issues

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

tai_alphi-2.0.1.tar.gz (33.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

tai_alphi-2.0.1-py3-none-any.whl (34.1 kB view details)

Uploaded Python 3

File details

Details for the file tai_alphi-2.0.1.tar.gz.

File metadata

  • Download URL: tai_alphi-2.0.1.tar.gz
  • Upload date:
  • Size: 33.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.2.1 CPython/3.10.9 Linux/6.11.0-1018-azure

File hashes

Hashes for tai_alphi-2.0.1.tar.gz
Algorithm Hash digest
SHA256 1a4489b31db83553a0d810f34eea6b5034acd40685c62b0f5ded3a67b5c79dac
MD5 1feda6d0520114628223f69e39969c0c
BLAKE2b-256 a76e133d9ddb3b4f57cdd318e86fb71c850c4b507499f3217483393cb5f60a86

See more details on using hashes here.

File details

Details for the file tai_alphi-2.0.1-py3-none-any.whl.

File metadata

  • Download URL: tai_alphi-2.0.1-py3-none-any.whl
  • Upload date:
  • Size: 34.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.2.1 CPython/3.10.9 Linux/6.11.0-1018-azure

File hashes

Hashes for tai_alphi-2.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 7964f03d5f5233902049b7df258293ad98117e3feec64f1a27d96857ecbc0789
MD5 5b436570e02d5ab306403ca49a304831
BLAKE2b-256 dabf03131d41bc020915b552738780a275e932a7a3124f56a11ea02fa199ef9c

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page