devstream-client 0.3.0

Python client library for DevStream logging service

DevStream Python Client

Python client library for sending logs to DevStream logging service.

For AI coding agents: a machine-readable integration guide is hosted at your DevStream server's /llms.txt (e.g. https://devstream.fly.dev/llms.txt), covering this client, the raw HTTP API, and the OpenAPI spec.

Installation

pip install devstream-client

Or install from source:

cd client
pip install -e .

Quick Start

Basic Usage

from devstream_client import DevStreamClient

# Initialize client
client = DevStreamClient(
    api_key="your-api-key",
    app_key="my-app",
    deployment_key="production",
    base_url="http://localhost:8787"  # or your production URL
)

# Send a simple log
client.log("Application started successfully")

# Send a log with tags
client.log_with_tags(
    "User logged in",
    user_id="12345",
    level="info",
    ip_address="192.168.1.1"
)

# Attach structured JSON data to a log
client.log(
    "Order placed",
    tags=[{"name": "level", "value": "info"}],
    data={"order_id": 4823, "total": 19.99, "items": ["sku-1", "sku-2"]},
)

# data also works with the keyword-tag helper
client.log_with_tags(
    "Order placed",
    data={"order_id": 4823, "total": 19.99},
    level="info",
)

Batch Sending

Send many messages in a single HTTP request (one bulk insert server-side):

client.log_batch([
    {"message": "step 1", "correlation_id": "req-1"},
    {"message": "step 2", "tags": [{"name": "level", "value": "info"}]},
    {"message": "step 3", "data": {"rows": 42}},
])

Each entry accepts message (required), tags, data, and correlation_id.

Python Logging Integration

import logging
from devstream_client import DevStreamHandler

# Create logger
logger = logging.getLogger("myapp")
logger.setLevel(logging.INFO)

# Add DevStream handler
handler = DevStreamHandler(
    api_key="your-api-key",
    app_key="my-app",
    deployment_key="production",
    base_url="http://localhost:8787"  # or your production URL
)
logger.addHandler(handler)

# Use standard Python logging
logger.info("Application started")
logger.error("Something went wrong", exc_info=True)

The handler is non-blocking: each record is queued and sent on a background thread, so logging never blocks the calling thread on the network (important under async servers like Daphne and in Celery tasks). If the queue fills up (default 1000 pending records, configurable via queue_size), new records are dropped rather than blocking. Pending records are flushed on interpreter exit.

For high-volume logging, pass batch=True to coalesce queued records into a single batch request per drain (uses the server's batch endpoint):

handler = DevStreamHandler(..., batch=True, batch_max=100)

Context tags (correlation id, etc.)

Attach request-scoped context to every log line as a tag, without custom handler code, using a filter that reads a contextvars.ContextVar:

import contextvars
from devstream_client import ContextVarTagFilter

slack_thread = contextvars.ContextVar("slack_thread", default=None)
handler.addFilter(ContextVarTagFilter(slack_thread, "slack_thread"))

slack_thread.set("T123.456")  # set per request/task; appears as a tag on logs

If you use asgi-correlation-id, there's a ready-made filter (install with pip install devstream-client[asgi]):

from devstream_client import CorrelationIdFilter

handler.addFilter(CorrelationIdFilter())  # adds a "correlation_id" tag

Django Integration

Add to your Django settings:

# settings.py

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'devstream': {
            'class': 'devstream_client.client.DevStreamHandler',
            'api_key': 'your-api-key',
            'app_key': 'my-django-app',
            'deployment_key': os.environ.get('DEPLOYMENT', 'local'),
            'base_url': 'http://localhost:8787',  # or your production URL
        },
    },
    'loggers': {
        'django': {
            'handlers': ['devstream'],
            'level': 'INFO',
        },
        'myapp': {
            'handlers': ['devstream'],
            'level': 'DEBUG',
        },
    },
}

Configuration

Client Parameters

• api_key (required): Your DevStream API key
• app_key (required): Application identifier
• deployment_key (required): Deployment environment (e.g., 'local', 'dev', 'staging', 'production')
• base_url (optional): DevStream service URL (default: http://localhost:8787)
• timeout (optional): Per-request timeout in seconds (default: 5.0)
• max_retries (optional): Automatic retries on connection errors and transient 5xx responses (default: 2)

Structured Data

Pass a data dictionary to attach arbitrary JSON to a log message. It is stored alongside the message and returned by the API and log viewer:

client.log("Payment failed", data={"amount": 50, "currency": "USD", "code": "card_declined"})

Tags

Tags are key-value pairs that help you filter and search logs. You can add tags in two ways:

1. As a list of dictionaries:

client.log("User action", tags=[
    {"name": "user_id", "value": "12345"},
    {"name": "action", "value": "login"}
])

2. As keyword arguments:

client.log_with_tags(
    "User action",
    user_id="12345",
    action="login"
)

Features

• Simple API for sending logs
• Integration with Python's standard logging module
• Support for custom tags
• Automatic retry and error handling
• Django-ready configuration

License

MIT License