SEER SDK & Libraries

Python SDK (seerpy)

The official Python library for integrating Seer monitoring into your scripts and applications.

pip install seerpy

REST API

Use our REST API for any programming language or platform.

1. Installation

# Install the Python SDK
pip install seerpy

# Or use the REST API directly
curl -X POST https://api.ansrstudio.com/monitoring \
  -H "Authorization: your_api_key" \
  -H "Content-Type: application/json"

2. Basic Usage

from seerpy import Seer
import os

# api_key= preferred; apiKey= kept for backwards compatibility
seer = Seer(api_key=os.getenv("SEER_API_KEY"))

# job_name must match a pipeline in your Seer dashboard
with seer.monitor("my_pipeline", capture_logs=True):
    # Monitoring never raises — Seer outages cannot fail your job
    process_data()

3. Advanced Features

from seerpy import Seer
import os

seer = Seer(
    api_key=os.getenv("SEER_API_KEY"),
    auto_replay=True,          # flush offline queue once on init
    background_replay=True,    # daemon thread flushes periodically
    replay_interval=60,        # seconds between background flushes
)

with seer.monitor(
    "data_pipeline",
    capture_logs=True,
    metadata={"environment": "production", "version": "1.2.0"},
    tags=["etl", "prod"],
):
    process_data()
    seer.heartbeat("data_pipeline", metadata={"progress": "50%"})

Core Features

• Automatic Error Handling: Captures exceptions and stack traces
• Progress Logging: Log checkpoints and intermediate results
• Metadata Support: Attach custom data to runs
• Heartbeat Monitoring: Track long-running processes
• Offline Mode: Queue API calls when network is unavailable
• Retry Logic: Automatic retry for failed API calls

Configuration Options

from seerpy import Seer
import os

seer = Seer(
    api_key=os.getenv("SEER_API_KEY"),
    base_url="https://api.ansrstudio.com",  # or SEER_BASE_URL env var
    timeout=30,                             # HTTP timeout in seconds
    auto_replay=True,                       # flush queue on init
    background_replay=True,                 # daemon thread flush
    replay_interval=60,                     # seconds between flushes
)

Context Manager

Use Python's context manager for automatic cleanup:

with seer.monitor(job_name="data-pipeline"):
    process_data()
    # Automatically calls seer.success() on exit
    # Calls seer.failure() on exception

Offline Replay

Flush the offline queue manually or automatically:

# Manual flush anytime
result = seer.replay()
print(result.sent, result.failed, result.dead_lettered)

# Stop the background replay daemon (optional clean shutdown)
seer.stop_background_replay()

Batch Operations

Monitor batch processing with heartbeats:

with seer.monitor("batch_processor", capture_logs=True):
    for i, batch in enumerate(batches):
        process_batch(batch)
        seer.heartbeat(
            "batch_processor",
            metadata={"batch": i + 1, "total": len(batches)},
        )