Python

pip install streamlogia

import os
from logingestor import LogIngestorClient

client = LogIngestorClient(
    api_key=os.environ["LOGINGESTOR_API_KEY"],
    project_id=os.environ["LOGINGESTOR_PROJECT_ID"],
    source="my-service",
)

client.debug("Cache miss", meta={"key": "user:abc"})
client.info("Order created", meta={"order_id": "ord_99"}, tags=["orders"])
client.warn("Rate limit approaching", tags=["alerts"])
client.error("Payment failed", meta={"order_id": "o_456", "reason": "card_declined"})

client.close()  # flush remaining logs before exit

import logging
import os
from logingestor import LogIngestorClient

client = LogIngestorClient(
    api_key=os.environ["LOGINGESTOR_API_KEY"],
    project_id=os.environ["LOGINGESTOR_PROJECT_ID"],
    source="my-service",
)

logger = logging.getLogger("my-service")
logger.setLevel(logging.DEBUG)
logger.addHandler(client.logging_handler())

logger.info("Server started on port 8080")
logger.error("Database connection failed", exc_info=True)
# ↑ Exception traceback is automatically captured in meta

logger.info("payment record persisted", extra={"meta": {"payment_id": "pay_abc"}})

import os
import signal
import sys
from flask import Flask, jsonify, request
from logingestor import LogIngestorClient

client = LogIngestorClient(
    api_key=os.environ["LOGINGESTOR_API_KEY"],
    project_id=os.environ["LOGINGESTOR_PROJECT_ID"],
    source="payment-service",
)

signal.signal(signal.SIGTERM, lambda *_: (client.close(), sys.exit(0)))

app = Flask(__name__)
client.flask_middleware(app)  # logs every request automatically

@app.post("/payments")
def create_payment():
    data = request.get_json()
    payment_id = f"pay_{os.urandom(4).hex()}"
    client.info("payment processed", meta={
        "payment_id": payment_id,
        "customer_id": data.get("customerId"),
        "amount": data.get("amount"),
    }, tags=["payments"])
    return jsonify({"paymentId": payment_id}), 201

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from logingestor import LogIngestorClient

client = LogIngestorClient(
    api_key=os.environ["LOGINGESTOR_API_KEY"],
    project_id=os.environ["LOGINGESTOR_PROJECT_ID"],
    source="order-service",
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    client.close()

app = FastAPI(lifespan=lifespan)
app.add_middleware(client.asgi_middleware())  # logs every request automatically

@app.post("/orders", status_code=201)
async def create_order(body: dict):
    order_id = f"ord_{os.urandom(4).hex()}"
    client.info("order created", meta={
        "order_id": order_id,
        "customer_id": body.get("customerId"),
    }, tags=["orders"])
    return {"orderId": order_id}

import os
import urllib.request
import json

def ingest(entries: list) -> None:
    body = json.dumps(entries).encode()
    req = urllib.request.Request(
        "https://api.streamlogia.com/v1/ingest",
        data=body,
        headers={
            "Authorization": f"Bearer {os.environ['LOGINGESTOR_API_KEY']}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=5) as resp:
        if resp.status >= 400:
            raise RuntimeError(f"ingest failed with status {resp.status}")

ingest([{
    "projectId": os.environ["LOGINGESTOR_PROJECT_ID"],
    "level":     "INFO",
    "message":   "Hello, Streamlogia!",
    "source":    "my-service",
    "timestamp": "2024-01-01T00:00:00Z",
    "tags":      [],
    "meta":      {},
}])