aepipe-sdk

aepipe-sdk-python

Python SDK for aepipe, a multi-tenant log ingestion and query service built on Cloudflare Workers Analytics Engine.

Zero external dependencies. Python 3.10+.

Installation

pip install aepipe-sdk

Quick Start

from aepipe import Aepipe

client = Aepipe(
    base_url="https://aepipe.yourdomain.com",
    token="your-admin-token",
)

API Reference

ingest(project, logstore, points) -> IngestResult

Write structured event points to Analytics Engine.

from aepipe import DataPoint

result = client.ingest("myapp", "backend", [
    DataPoint(event="user_login"),
    DataPoint(event="api_error", level="error", blobs=["GET /api"], doubles=[1.23]),
    # With D1 payload — large data stored separately, linked via UUID
    DataPoint(
        event="unhandled_exception",
        level="error",
        payload={"stack": "Error: ...\n  at ...", "request": {"url": "/api/users"}},
        ttl=604800,  # 7 days
    ),
])
print(result.ok)       # True
print(result.written)  # 3

log(project, logstore, logs) -> LogResult

Write raw log entries via Workers Observability.

from aepipe import LogEntry

result = client.log("myapp", "backend", [
    LogEntry(message="server started"),
    LogEntry(message="connection timeout", level="error", extra={"ip": "1.2.3.4"}),
])
print(result.written)  # 2

query(project, logstore, sql) -> QueryResult

Run a SQL query against Analytics Engine. Tenant filters are injected automatically.

result = client.query("myapp", "backend", "SELECT count() as cnt FROM aepipe")
print(result.data)

rawlog(project, logstore, *, limit=50, start=None, end=None) -> RawLogResult

Query raw Worker logs via Cloudflare Telemetry API.

result = client.rawlog("myapp", "backend", limit=100, start="2025-01-01T00:00:00Z")
for entry in result.logs:
    print(f"[{entry.level}] {entry.timestamp} {entry.data}")
print(f"total: {result.count}")

detail(project, logstore, ref_ids) -> DetailResult

Fetch extended payloads from D1 by ref_id. Requires D1 binding on the server.

# 1. Query AE to get ref_ids
ae_result = client.query("myapp", "backend",
    "SELECT blob5 as ref_id, blob3 as event FROM aepipe WHERE blob5 != '' LIMIT 10"
)

# 2. Fetch full payloads from D1
ref_ids = [r["ref_id"] for r in ae_result.data.get("data", [])]
details = client.detail("myapp", "backend", ref_ids)
for entry in details.results:
    print(entry.ref_id, entry.payload)  # {"stack": "...", "request": {...}}

list_projects() -> ListResult

List all projects that have written data.

result = client.list_projects()
print(result.items)  # ["myapp", "analytics", ...]

list_logstores(project) -> ListResult

List all logstores within a project.

result = client.list_logstores("myapp")
print(result.items)  # ["backend", "frontend", ...]

Data Types

DataPoint

blobs vs doubles — how to choose?

• blobs (string): Used for grouping (GROUP BY) and filtering (WHERE). Not aggregatable. Examples: URL path, user ID, region, error message.
• doubles (number): 64-bit IEEE 754 floats. Can be aggregated with SUM, AVG, QUANTILE, MIN, MAX. Examples: response time, request size, error count.

Cloudflare Analytics Engine Limits

These are hard limits enforced by Cloudflare. Exceeding blob size limits causes silent data truncation — the write succeeds but data is lost, making it impossible to observe anomalies. The SDK validates these limits client-side and raises ValidationError before sending.

Tip: If your data is large (e.g., full stack traces, request dumps), use the payload field instead of blobs. Payloads are stored in Cloudflare D1 with no 16 KB limit, linked to AE via a UUID in blob5.

See: Cloudflare Analytics Engine Limits

LogEntry

Result Types

Validation

All methods validate inputs before making network requests:

• Name format: project and logstore names must match ^[a-zA-Z0-9_-]{1,64}$.
• Batch size: ingest() and log() accept at most 250 items per call.

Invalid inputs raise ValidationError (subclass of ValueError).

Error Handling

from aepipe import AepipeError, ValidationError

try:
    client.ingest("myapp", "backend", points)
except ValidationError as e:
    # Client-side validation failed
    print(f"invalid input: {e}")
except AepipeError as e:
    # Server returned an error
    print(f"API error {e.status}: {e.message}")

License

MIT