datares-logger 0.1.0

Official Python SDK for Datares Logger — structured log ingestion by Datares

Datares Logger Python SDK

Official Python SDK for Datares Logger — structured log ingestion by Datares.

Requirements: Python 3.9+

---

Installation

pip install datares-logger

---

Quick Start

from datares_logger import Logger

logger = Logger("dtr_live_<your-api-key>")

logger.info("Application started", service="api")
logger.warning("Disk usage high", service="api", meta={"usage_pct": 92})
logger.error("Payment failed", service="billing", meta={"order_id": "ord_123"})

# The buffer is flushed automatically when the process exits.
# Call flush() explicitly if you need to ensure delivery before that point.
logger.flush()

---

Logger

Constructor

Logger(api_key: str, config: Config = Config())

Parameter	Type	Description
api_key	str	Datares API key (dtr_live_<48 hex chars>)
config	Config	Optional SDK configuration (see below)

Convenience Methods

All convenience methods accept the same arguments:

logger.debug   (message, service="default", meta={})
logger.info    (message, service="default", meta={})
logger.warning (message, service="default", meta={})
logger.error   (message, service="default", meta={})
logger.critical(message, service="default", meta={})

Parameter	Type	Description
message	str	Human-readable log message (required)
service	str	Service / component name (default: "default")
meta	dict	Arbitrary key-value metadata

These methods add the entry to the internal buffer. The buffer is flushed automatically when it reaches config.batch_size or when the process exits.

log(entry: LogEntry)

Add a pre-built LogEntry to the buffer.

from datares_logger import Logger, LogEntry

logger = Logger("dtr_live_...")
entry = LogEntry(level="info", message="Custom entry", service="worker")
logger.log(entry)

send(entries) -> int

Send one or more entries immediately, bypassing the buffer. Raises on any failure (does not call the error handler).

count = logger.send(LogEntry(level="error", message="Critical alert"))
count = logger.send([entry1, entry2, entry3])

Returns the number of entries accepted by the server.

flush()

Flush all buffered entries to the API. On failure the registered error handler is called (default: print to stderr). The buffer is always cleared, regardless of success or failure.

logger.flush()

on_error(handler) -> Logger

Register a callback invoked when flush() fails. Returns self for method chaining.

import sentry_sdk

logger.on_error(lambda err: sentry_sdk.capture_exception(err))

---

LogEntry

from datares_logger import LogEntry
from datetime import datetime, timezone

entry = LogEntry(
    level="warning",
    message="Something went wrong",
    service="payments",
    timestamp=datetime(2024, 6, 15, 12, 0, 0, tzinfo=timezone.utc),
    meta={"order_id": "ord_456", "amount": 9900},
)

Field	Type	Required	Default	Description
level	str	Yes	—	Log level string (see Level enum)
message	str	Yes	—	Log message
service	str	No	"default"	Service / component name
timestamp	datetime or None	No	current time	Event time (RFC 3339 when serialised)
meta	dict	No	{}	Arbitrary key-value metadata

---

Level

from datares_logger import Level

Level.DEBUG    # "debug"
Level.INFO     # "info"
Level.WARNING  # "warning"
Level.ERROR    # "error"
Level.CRITICAL # "critical"

---

Config

from datares_logger import Config, Logger

config = Config(
    base_url="https://api.datares.id",
    timeout=10,
    retries=3,
    batch_size=100,
    auto_flush=True,
)

logger = Logger("dtr_live_...", config=config)

Option	Type	Default	Description
base_url	str	"https://api.datares.id"	API base URL (no trailing slash)
timeout	int	10	HTTP request timeout in seconds
retries	int	3	Retry attempts on 5xx errors (exponential back-off)
batch_size	int	100	Auto-flush buffer when this many entries are buffered (1–500)
auto_flush	bool	True	Register atexit handler to flush on process exit

---

Error Handling

from datares_logger import Logger, AuthError, RateLimitError, ApiError, LoggerError

logger = Logger("dtr_live_...")

# send() raises — handle explicitly:
try:
    logger.send(entry)
except AuthError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ApiError as e:
    print(f"API error: {e} (HTTP {e.status_code})")

# flush() uses the error handler:
logger.on_error(lambda err: print(f"Flush failed: {err}"))
logger.flush()

Exception Hierarchy

LoggerError
├── AuthError          — HTTP 401 / 403 (invalid or missing API key)
├── RateLimitError     — HTTP 429  (.retry_after: Optional[int])
└── ApiError           — HTTP 422, 5xx after retries  (.status_code: int)

---

Framework Integration

Django

In settings.py or your app AppConfig.ready():

from datares_logger import Logger, Config

datares = Logger(
    api_key="dtr_live_...",
    config=Config(batch_size=50),
)

In views / signals:

datares.info("User signed in", service="accounts", meta={"user_id": request.user.id})

Flask

from flask import Flask, g
from datares_logger import Logger

app = Flask(__name__)
datares = Logger("dtr_live_...")

@app.before_request
def log_request():
    datares.info("Incoming request", service="web", meta={"path": request.path})

Standalone Script

from datares_logger import Logger

logger = Logger("dtr_live_...")

def main():
    logger.info("Script started", service="cronjob")
    # ... do work ...
    logger.info("Script finished", service="cronjob")
    # flush() is called automatically by atexit

if __name__ == "__main__":
    main()

---

Running Tests

pip install -e ".[dev]"
pytest

# With JUnit XML report:
pytest --junit-xml=junit.xml

---

License

MIT — see LICENSE.