Metadata-Version: 2.4
Name: loguru-seq
Version: 0.1.0
Summary: A loguru sink that sends structured logs to Seq using CLEF format
Project-URL: Repository, https://github.com/gabrielgt/loguru-seq
License: MIT
License-File: LICENSE
Keywords: clef,logging,loguru,seq,sink
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: System :: Logging
Requires-Python: >=3.9
Requires-Dist: httpx>=0.27
Requires-Dist: loguru>=0.6
Description-Content-Type: text/markdown

# loguru-seq

A [loguru](https://github.com/Delgan/loguru) sink that sends structured logs to [Seq](https://datalust.co/seq) using the [CLEF](https://clef-json.org/) format over HTTP with background batching.

## Features

- Non-blocking background thread with configurable batch interval and size
- Structured fields via loguru's `bind()` forwarded as CLEF properties
- Exception traces forwarded as the CLEF `@x` field
- Optional API key authentication
- Graceful shutdown that flushes remaining events on process exit

## Installation

```bash
pip install loguru-seq
```

Or with [uv](https://github.com/astral-sh/uv):

```bash
uv add loguru-seq
```

## Quick start

```python
from loguru import logger
from loguru_seq import SeqSink

sink = SeqSink(url="http://localhost:5341")
logger.add(sink)

logger.info("Application started")
logger.warning("Low disk space")
```

### With an API key

```python
sink = SeqSink(
    url="http://seq.example.com",
    api_key="your-api-key",
)
logger.add(sink)
```

### Structured properties with `bind()`

Extra fields passed to `logger.bind()` are forwarded as top-level CLEF properties and become searchable in Seq.

```python
logger.bind(user_id=42, order_id="ORD-999").info("Order placed")
logger.bind(component="auth").warning("Token about to expire")
```

### Structured properties via message arguments

Named keyword arguments passed directly to the log call are also forwarded as CLEF properties, and additionally rendered into the message text:

```python
logger.info("User {name} logged in from {ip}", name="alice", ip="192.168.1.1")
# @mt  → "User alice logged in from 192.168.1.1"
# name → "alice"   (searchable property in Seq)
# ip   → "192.168.1.1"  (searchable property in Seq)
```

The difference from `bind()` is that the values are **also baked into `@mt`**, so they appear twice: in the message text and as discrete properties. With `bind()` the message text stays as a clean template and the values only live in their respective properties.

Positional arguments (e.g. `logger.info("value is {}", 42)`) are used for rendering only and do **not** become CLEF properties.

Use `bind()` for context that applies to multiple log calls (e.g. a request ID). Use inline kwargs when the value is specific to a single message and you want it visible in the message text as well.

### Exception capture

```python
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception("Unexpected error in calculation")
```

The full traceback is sent in the CLEF `@x` field and displayed in Seq's exception viewer.

## Configuration reference

```python
SeqSink(
    url,                          # Base URL of the Seq server, e.g. "http://localhost:5341"
    api_key=None,                 # Seq API key (Settings → API Keys in the Seq UI)
    batch_interval_seconds=5.0,   # Seconds between automatic flushes
    batch_size=100,               # Flush early when this many events are buffered
    timeout_seconds=10.0,         # HTTP request timeout, in seconds
    max_queue_size=10_000,        # Drop events (with a stderr warning) when queue exceeds this
)
```

| Parameter | Default | Description |
|---|---|---|
| `url` | — | Base URL of the Seq instance |
| `api_key` | `None` | API key for authenticated Seq instances |
| `batch_interval_seconds` | `5.0` | Maximum seconds to wait before flushing |
| `batch_size` | `100` | Flush immediately when this many events are queued |
| `timeout_seconds` | `10.0` | HTTP timeout, in seconds, for each flush request |
| `max_queue_size` | `10_000` | Hard cap on the in-memory event queue |

### Manual flush

Call `sink.flush()` to send all buffered events immediately, for example before a scheduled job ends.

```python
sink = SeqSink(url="http://localhost:5341", batch_interval_seconds=60)
logger.add(sink)

logger.info("Job started")
# ... work ...
logger.info("Job finished")
sink.flush()
```

## Log level mapping

| loguru | CLEF / Seq |
|---|---|
| `TRACE` | `Verbose` |
| `DEBUG` | `Debug` |
| `INFO` | `Information` |
| `SUCCESS` | `Information` |
| `WARNING` | `Warning` |
| `ERROR` | `Error` |
| `CRITICAL` | `Fatal` |

---

## Contributing

Contributions are welcome. Please open an issue before submitting large changes so we can discuss approach first.

### Prerequisites

- Python 3.9 or later
- [uv](https://github.com/astral-sh/uv) — used for environment and dependency management

### Setting up the development environment

```bash
# Clone the repository
git clone https://github.com/gabrielgt/loguru-seq.git
cd loguru-seq

# Create the virtual environment and install all dependencies (including dev extras)
uv sync
```

`uv sync` reads [pyproject.toml](pyproject.toml) and [uv.lock](uv.lock), creates a `.venv` directory, and installs the package in editable mode along with the `dev` dependency group (`pytest`, `pytest-mock`, `ruff`, `pyright`).

Alternatively, open the repository in the provided [dev container](.devcontainer/devcontainer.json) (VS Code's "Reopen in Container", or GitHub Codespaces) for a preconfigured Python 3.11 environment with `uv` and the recommended VS Code extensions already installed.

### Project structure

```
src/loguru_seq/
    __init__.py     # Public API — exports SeqSink
    sink.py         # SeqSink: buffering, batching, background thread
    sender.py       # SeqSender: httpx client, sends CLEF to Seq's /api/events/raw
    clef.py         # loguru record → CLEF dict, list of dicts → NDJSON string
tests/
    unit/           # Fast, no network required
    integration/    # Require a running Seq instance
```

### Running unit tests

Unit tests use mocks and require no external services.

```bash
uv run pytest tests/unit
```

### Running integration tests

Integration tests send real HTTP requests to a Seq instance.

1. Start a local Seq instance (Docker is the easiest way):

```bash
docker run --name seq -d --restart unless-stopped \
  -e ACCEPT_EULA=Y \
  -e SEQ_PASSWORD=<initial-password> \
  -p 5341:80 \
  datalust/seq:latest
```

2. Copy the example environment file and fill in your values:

```bash
cp example.env .env
# Edit .env — set SEQ_URL and optionally SEQ_API_KEY
```

The defaults in `.env` point to `http://host.docker.internal:5341`, which resolves correctly from inside another Docker container (for example a dev container). If you are running tests directly on your host machine, change `SEQ_URL` to `http://localhost:5341`.

3. Run the integration tests:

```bash
uv run pytest tests/integration
```

### Running all tests

```bash
uv run pytest
```

### Code style

This project uses [ruff](https://docs.astral.sh/ruff/) for linting/formatting and [pyright](https://github.com/microsoft/pyright) for type checking. All public functions and methods should be fully type-annotated.

Run everything CI checks (lint, format, type check, unit tests, integration tests) with one command:

```bash
./scripts/check.sh
```

Integration tests skip automatically with a clear message if no Seq instance is reachable — see [Running integration tests](#running-integration-tests) to set one up locally.

Or run the checks individually:

```bash
uv run ruff check .            # Lint
uv run ruff format --check .   # Format (check only; use `ruff format .` to apply)
uv run pyright                  # Type check (checks src/ only)
uv run pytest tests/unit        # Unit tests
uv run pytest tests/integration # Integration tests (skipped if Seq isn't reachable)
```

### Submitting changes

1. Fork the repository and create a branch from `main`.
2. Make your changes with tests.
3. Run `./scripts/check.sh` and verify it passes (this mirrors CI's `test` job; the `integration` job runs the same `tests/integration` suite against a real Seq service container in CI).
4. Open a pull request with a clear description of what and why.

---

## License

MIT
