Loki Console Logger Python

Elven Observability Official Logger:

Um coletor de logs, eventos e exceções de apps Python que envia tudo para o Grafana Loki, com suporte a labels dinâmicas, alta performance, envio assíncrono e integração instantânea com o Elven Stack.

Features

• Altíssima performance com asyncio + aiohttp

• Captura tudo automaticamente:

  • print()

  • logging

  • Exceções não tratadas

  • Eventos personalizados via track_event

• Suporte a labels fixas e dinâmicas

• Buffer com envio em lote e intervalo configurável

• Seguro contra falhas de rede

• Compatível com apps síncronos e assíncronos

• Envio final garantido com atexit (até em scripts simples)

Instalação

pip install loki-console-logger-python

Uso básico

from loki_console_logger_python import LokiLogger
from loki_console_logger_python.config import LokiLoggerOptions

options = LokiLoggerOptions(
    url="<https://loki.elvenobservability.com>",
    tenant_id="elven",
    app_name="my-awesome-app",
    auth_token="seu-token-opcional",  # JWT Elven
    batch_size=10,
    flush_interval=2,
    labels={"env": "production"},
    dynamic_labels={"hostname": lambda: "api-01.local"},
)

logger = LokiLogger(options)

print("Isso será enviado ao Loki!")
logger.track_event("usuario_cadastrado", {"user_id": 123})
raise Exception("Erro de teste!")  # Capturado automaticamente

Opções de Configuração

Parâmetro	Tipo	Descrição	Padrão
url	str	Endpoint do Loki HTTP Push API	—
tenant_id	str	Usado no header X-Scope-OrgID (multi-tenant)	—
app_name	str	Label de identificação da aplicação	—
auth_token	str	Token JWT para autenticação	None
batch_size	int	Quantidade de logs para envio em lote	10
flush_interval	int (segundos)	Tempo máximo antes do envio	2
labels	dict[str, str]	Labels fixas	{}
dynamic_labels	dict[str, Callable[[]])	Labels geradas dinamicamente em cada envio	{}

Exemplo com FastAPI

from fastapi import FastAPI
from loki_console_logger_python import LokiLogger
from loki_console_logger_python.config import LokiLoggerOptions

app = FastAPI()

logger = LokiLogger(
    LokiLoggerOptions(
        url="<https://loki.elvenobservability.com>",
        tenant_id="elven",
        app_name="fastapi-app",
        auth_token="seu-token",
        labels={"env": "dev"},
        dynamic_labels={"hostname": lambda: "devbox"}
    )
)

@app.get("/ping")
async def ping():
    print("ping chamado")
    logger.track_event("ping")
    return {"message": "pong"}

@app.get("/erro")
async def erro():
    raise ValueError("Erro proposital")

Boas práticas

Situação	Recomendações
Script curto (ex: CLI, cron)	O logger garante envio com atexit, mas prefira logger.flush_sync() no final para garantir.
Aplicações web (ex: FastAPI)	Nada extra é necessário, o envio é feito automaticamente em background.
Ambientes sem asyncio rodando	Não se preocupe: o logger detecta e adapta automaticamente
Alta concorrência	Use batch_size > 10 e ajuste flush_interval para evitar excesso de requisições

Troubleshooting

Problema	Causa comum	Solução
SyntaxError em linha com url=	Caractere invisível (ex: vindo de copy/paste)	Reescreva o trecho manualmente
Logs não aparecem no Loki	Labels inconsistentes ou sem flush suficiente	Verifique tenant_id, auth_token, use flush_logs() antes de encerrar
RuntimeError em atexit	Script puro sem loop asyncio	Já tratado internamente com fallback seguro
Uvicorn reiniciando em loop	Código com erro de sintaxe	Veja o log e corrija a linha apontada (normalmente url=)

Flush manual (opcional)

Em scripts curtos ou síncronos, você pode forçar o envio assim:

logger.flush_sync()  # Bloca até todos os logs serem enviados

Eventos customizados

logger.track_event("usuario_ativo", {"user_id": "abc123", "source": "mobile"})

Esses eventos aparecerão no Loki como:

[EVENT] usuario_ativo {"user_id": "abc123", "source": "mobile"}

Formato dos logs no Loki

Cada entrada tem:

• Timestamp com precisão de nanossegundos

• Labels definidas na config

• Mensagem prefixada com [PRINT], [EXCEPTION], [EVENT], [INFO], etc

Licença

Este projeto é licenciado sob a MIT License.

Contribuindo

1. Fork este repositório

2. Crie uma branch com sua feature ou correção

3. Envie um Pull Request

Toda contribuição é bem-vinda para tornar a observabilidade Python ainda mais poderosa!

Feito com amor pela Elven Observability

Ficou com dúvida? Fale com a gente!