Loki Console Logger Python

Elven Observability Official Logger: A log, event, and exception collector for Python apps that sends everything to Grafana Loki, with support for dynamic labels, high performance, asynchronous delivery, and instant integration with the Elven Stack.

Features

• Extremely high performance with asyncio + aiohttp

• Automatically captures everything:

  • print()

  • logging

  • Unhandled exceptions

  • Custom events via track_event

• Support for static and dynamic labels

• Buffer with batch delivery and configurable interval

• Resilient to network failures

• Compatible with synchronous and asynchronous apps

• Guaranteed final delivery with atexit (even in simple scripts)

Installation

pip install loki-console-logger-python

Basic usage

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="your-optional-token",  # Elven JWT
    batch_size=10,
    flush_interval=2,
    labels={"env": "production"},
    dynamic_labels={"hostname": lambda: "api-01.local"},
)

logger = LokiLogger(options)

print("This will be sent to Loki!")
logger.track_event("registered_user", {"user_id": 123})
raise Exception("Test error!")  # Automatically captured

Configuration Options

Parameter	Type	Description	Default
url	str	Loki HTTP Push API endpoint	—
tenant_id	str	Used in the X-Scope-OrgID header (multi-tenant)	—
app_name	str	Application identification label	—
auth_token	str	JWT token for authentication	None
batch_size	int	Number of logs for batch sending	10
flush_interval	int (segundos)	Maximum time before sending	2
labels	dict[str, str]	Fixed labels	{}
dynamic_labels	dict[str, Callable[[]])	Labels generated dynamically with each submission	{}

Example with 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="your-token",
        labels={"env": "dev"},
        dynamic_labels={"hostname": lambda: "devbox"}
    )
)

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

@app.get("/erro")
async def erro():
    raise ValueError("Intentional error")

Best practices

Situation	Recommendations
Short script (e.g., CLI, cron)	The logger ensures delivery with atexit,but preferlogger.flush_sync()at the end to guarantee it.
Web applications (e.g., FastAPI)	Nothing extra is required, delivery is handled automatically in the background.
Environments without asyncio running	Don't worry: the logger detects and adapts automatically.
High concurrency	Use batch_size > 10 and adjust flush_interval to avoid excessive requests.

Troubleshooting

Problem	Common cause	Solution
SyntaxError on line withurl=	Invisible character (e.g., from copy/paste)	Rewrite the section manual.y
Logs are not showing up in Loki.	Inconsistent labels or insufficient flushing	Check tenant_id, auth_token, use flush_logs() before exiting
RuntimeError in atexit	Plain script without asyncio loop	Already handled internally with a safe fallback
Uvicorn restarting in a loop	Code with syntax error	Check the log and fix the indicated line (usually url=)

Manual flush (optional)

In short or synchronous scripts, you can force the send like this:

logger.flush_sync()  # Blocks until all logs are sent

Custom events

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

These events will appear in Loki as:

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

Log format in Loki

Each entry includes:

• Timestamp with nanosecond precision

• Labels defined in the config

• Message prefixed with [PRINT], [EXCEPTION], [EVENT], [INFO], etc.

License

This project is licensed under the MIT License.

Contributing

1. Fork this repository

2. Create a branch with your feature or fix

3. Submit a Pull Request

Every contribution is welcome to make Python observability even more powerful!

Made with love by Elven Observability

Need help? Talk to us!