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

Fork este repositório
Crie uma branch com sua feature ou correção
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!