APISense Python SDK

Lightweight, privacy-first API observability for Python backend applications.

---

Overview

The APISense Python SDK provides a strongly typed, production-grade telemetry delivery layer for collecting and streaming request execution metadata to the APISense observability platform.

---

Installation

Install the package directly from PyPI:

pip install apisense

Or install with FastAPI extras:

pip install "apisense[fastapi]"

Local Development Installation

For local development from the repository:

pip install -e .

---

Architecture & Telemetry Delivery

FastAPI / Starlette Application
             │
             ▼
   APISenseMiddleware (ASGI)
             │ (captures latency, method, endpoint, status_code)
             ▼
      TelemetryEvent (Schema-validated, privacy-enforced)
             │
             ▼
    TelemetryCollector (Enabled/Disabled check)
             │
             ▼
  BoundedTelemetryBuffer (Thread-safe FIFO, O(1), Drop-Newest on full)
             │
             ▼
  BackgroundTelemetryWorker (Periodic batch draining & flush loop)
             │
             ▼
     TelemetryHTTPTransport (httpx with connection pooling & retries)
             │
             ▼
  APISense Backend Ingestion API (POST /api/v1/ingest)

Non-Blocking & Reliability Principles

• Zero application blocking: Telemetry collection on the request path is a pure $O(1)$ memory insertion (~50ns). All networking runs asynchronously in a dedicated background worker task.
• Strictly bounded memory: The in-memory buffer enforces max_buffer_size to prevent unbounded memory growth.
• Drop-Newest on full: Under heavy traffic spikes or downstream backend downtime, new events are gracefully dropped without raising exceptions.
• FastAPI failure isolation: Telemetry capture errors or transport failures are isolated internally and can never modify, delay, or crash your API responses.

---

Developer Integration with FastAPI

Standard Integration (Recommended)

from contextlib import asynccontextmanager
from fastapi import FastAPI
from apisense import APISense, APISenseMiddleware

# 1. Initialize the client
apisense = APISense(
    api_key="aps_live_your_project_api_key_here",
    environment="production",
)


# 2. Manage worker lifecycle with FastAPI's modern lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with apisense:
        yield


# 3. Attach lifespan and middleware
app = FastAPI(lifespan=lifespan)
app.add_middleware(APISenseMiddleware, client=apisense)


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": "Alice"}

Alternatively, supply the API key via the APISENSE_API_KEY environment variable:

export APISENSE_API_KEY="aps_live_your_project_api_key_here"

apisense = APISense()

---

Configuration Reference

The SDK configuration is strongly typed and immutable (APISenseConfig):

Parameter	Type	Default	Description
api_key	str	Required	APISense project API key (must begin with aps_live_).
base_url	str	https://api.apisense.co.in	Ingestion endpoint URL.
environment	str	production	Deployment environment name (production, staging, development).
enabled	bool	True	Master switch to toggle telemetry collection on or off.
batch_size	int	100	Maximum number of events per ingestion batch.
flush_interval_seconds	float	5.0	Time in seconds before buffered events are flushed.
request_timeout_seconds	float	5.0	HTTP read/request timeout in seconds.
connect_timeout_seconds	float	2.0	HTTP connection timeout in seconds.
max_buffer_size	int	10000	In-memory event buffer capacity limit.
max_retries	int	3	Maximum retry attempts for transient transmission failures.
retry_backoff_factor	float	0.5	Exponential backoff multiplier for retries.
shutdown_timeout_seconds	float	3.0	Maximum time to wait for buffer flush during shutdown.
sampling_rate	float	1.0	Sampling rate between 0.0 (0%) and 1.0 (100%).
debug	bool	False	Enable diagnostic logging for troubleshooting.

Security & Secret Masking

• The API key is never displayed in plain text in repr(), str(), or log messages.
• Outbound requests use connection pooling and sanitized error logging.

---

Telemetry Contract & Privacy Boundary

APISense enforces a strict, privacy-by-design data model. Telemetry events only capture runtime observability metrics.

Canonical Event Schema (TelemetryEvent)

from datetime import UTC, datetime
from apisense import TelemetryEvent

event = TelemetryEvent(
    timestamp=datetime.now(UTC),
    method="GET",
    endpoint="/api/v1/users/{user_id}",
    status_code=200,
    latency_ms=14.5,
    request_id="req_01h8abc123",
    environment="production",
)

Privacy Guarantee

The TelemetryEvent data model strictly rejects all sensitive payload fields at the schema layer (extra="forbid"):

• ❌ No request bodies
• ❌ No response bodies
• ❌ No authorization headers / tokens
• ❌ No cookies
• ❌ No passwords or secret keys
• ❌ No database credentials
• ❌ No environment variables
• ❌ No arbitrary metadata dictionaries

---

Exceptions

The SDK uses a focused exception hierarchy rooted at APISenseError:

APISenseError
├── ConfigurationError   # Raised on invalid or missing configuration parameters
└── TelemetryError       # Raised on invalid telemetry event construction or validation