kyrontha-lens 0.1.0

Official Kyrontha Lens SDK — buffered async log shipping to lens.kyrontha.io

kyrontha-lens

Official Python SDK for Kyrontha Lens — buffered, async, fire-and-forget log shipping.

• Non-blocking. Logging never waits for the network. A daemon thread flushes batches every 5s or every 50 events.
• Doesn't crash your app. A Lens outage drops events after retry; your code keeps running.
• Drains on shutdown. Hooks atexit, SIGTERM, SIGINT so the last few seconds of logs aren't lost on container restart.
• Zero runtime dependencies. Uses stdlib urllib, json, threading, logging only.

Install

# Once published to PyPI:
pip install kyrontha-lens

# Until then (early access from our git repo):
pip install "git+https://dev.azure.com/jblamba/Kyrontha%20Lens/_git/Kyrontha%20Lens#subdirectory=sdk/python"

Python 3.9+ required.

Get an API key

In Kyrontha Lens, go to Connections → Connect via SDK, name your connection, and copy the API key shown on screen. The first event you ship will auto-promote the connection from "awaiting first event" to "connected".

Quick start

import os
from kyrontha_lens import KyronthaLogger

log = KyronthaLogger(
    api_key=os.environ["KYRONTHA_KEY"],
    source="checkout-api",
)

log.info("order placed", order_id=42, user_id="u-123")
log.error("payment failed", order_id=42, reason="declined")

# For short-lived scripts: drain before exit so the in-flight POST has time
# to complete. Long-running services don't need this — atexit handles it.
log.flush()

stdlib logging integration

Most idiomatic for any Python app already using logging:

import logging, os
from kyrontha_lens import KyronthaHandler

logging.basicConfig(level=logging.INFO, handlers=[
    logging.StreamHandler(),
    KyronthaHandler(api_key=os.environ["KYRONTHA_KEY"], source="checkout-api"),
])

log = logging.getLogger(__name__)
log.info("order placed", extra={"order_id": 42, "user_id": "u-123"})
log.error("payment failed", extra={"order_id": 42, "reason": "declined"})

The handler ships every record's extra={...} fields as structured meta. Standard LogRecord attributes (pathname, lineno, funcName, etc.) are stripped to avoid noise.

structlog integration

structlog renders to stdlib logging by default — so the same KyronthaHandler works:

import logging, os, structlog
from kyrontha_lens import KyronthaHandler

logging.basicConfig(level=logging.INFO, handlers=[
    logging.StreamHandler(),
    KyronthaHandler(api_key=os.environ["KYRONTHA_KEY"], source="checkout-api"),
])

structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.render_to_log_kwargs,        # forward as `extra=`
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
)

log = structlog.get_logger()
log.info("order placed", order_id=42, user_id="u-123")

Loguru integration

Loguru's logger.add(handler) accepts any logging.Handler:

import os
from loguru import logger
from kyrontha_lens import KyronthaHandler

logger.add(KyronthaHandler(api_key=os.environ["KYRONTHA_KEY"], source="checkout-api"))
logger.info("order placed", order_id=42)

API

KyronthaLogger(api_key, *, endpoint=None, source=None, ...)

kwarg	type	default	notes
api_key	str	(required)	From a Lens SDK connection
endpoint	str | None	https://lens.kyrontha.io/api/ingest	Override for self-hosted Lens
source	str | None	KYRONTHA_SOURCE env, or 'app'	Shows in the Source column of the Lens UI
flush_interval_s	float	5.0	Background flush cadence (seconds)
flush_batch_size	int	50	Buffer size that forces an immediate flush
max_retries	int	3	Per-batch, exponential backoff (250ms → 500ms → 1s)
on_error	Callable[[Exception], …]	prints to stderr	Pass lambda e: None to silence

Methods

Method	Notes
log(level, message, **meta)	Generic — level is any string, lower-cased before send
fatal/error/warn/info/debug	Convenience for the standard levels
flush(timeout=10.0)	Force-flush — call before short-lived scripts exit
close(timeout=10.0)	Stop the background worker and drain

How it works

1. Every log() call appends to an in-memory buffer — synchronous, no I/O.
2. A daemon thread flushes the buffer to POST /api/ingest every flush_interval_s. The buffer also flushes immediately when it hits flush_batch_size.
3. The Lens backend tags every event with the connection_id linked to your API key, so the View logs button on your connection card filters precisely.
4. On graceful shutdown (atexit, SIGTERM, SIGINT) the SDK drains the buffer one last time.

Failure modes

• Lens unreachable / 5xx: retried up to max_retries with exponential backoff, then dropped. Your on_error callback is invoked. The host app continues.
• API key revoked / 4xx: dropped immediately (retry won't help). Your on_error callback is invoked.
• Process killed with SIGKILL or os._exit(0): events still in the buffer are lost. Use log.flush() before exiting if you can't tolerate that.

License

Proprietary. © Kyrontha. Reach out for commercial use.