# 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")
```

## B**est practices**

| Situation                                  | Recommendations                                                                                      |
| ------------------------------------------ | ---------------------------------------------------------------------------------------------------- |
| Short script (e.g., CLI, cron)             | The logger ensures delivery with `atexit`,but prefer`logger.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` **r**unning | 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 with`url=` | 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](https://www.notion.so/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](https://elven.works/)

**Need help? Talk to us!**