seerpy

Python SDK

v0.2.4

Official Python SDK for SEER — monitor jobs, capture logs, record heartbeats, and reliably report execution results including when the network is down.

Installation

Install seerpy using pip

pip install seerpy

Requirements: Python 3.7 or higher. New in 0.2.x: adds the filelock dependency for cross-process offline queue locking.

Quick Start

from seerpy import Seer

# Initialize with your API key
# api_key= is preferred; apiKey= is kept for backwards compatibility
seer = Seer(api_key="YOUR_API_KEY")

# Your existing imports
import pandas as pd

def run_report():
    data = pd.read_csv("sales_data.csv")
    processed = data.groupby("region").sum()
    processed.to_csv("daily_report.csv")

# job_name must match a pipeline that exists in your Seer dashboard
with seer.monitor("daily_report_job", tags=["etl", "prod"]):
    run_report()

That's it! SEER automatically tracks execution time, success/failure status, and sends notifications if something goes wrong. Monitoring never raises — Seer outages cannot mask your job's exception.

API Reference

Seer(api_key, auto_replay=False, background_replay=False, replay_interval=60, base_url=None, timeout=30)

Initialize the SEER client

Parameters:

api_key (str, required) — Your SEER API key. apiKey= is accepted for backwards compatibility.
auto_replay (bool) — Flush the offline queue once on client init. Default: False.
background_replay (bool) — Start a daemon thread that flushes the queue periodically. Default: False.
replay_interval (int) — Seconds between background flushes. Default: 60.
base_url (str) — Override the default API host (https://api.ansrstudio.com/). Can also be set via SEER_BASE_URL.
timeout (int) — HTTP timeout in seconds. Default: 30.

Example:

from seerpy import Seer
import os

seer = Seer(
    api_key=os.environ["SEER_API_KEY"],
    auto_replay=True,
    background_replay=True,
    replay_interval=60,
)

seer.monitor(job_name, capture_logs=False, metadata=None, tags=None)

Context manager for monitoring script execution

Parameters:

job_name (str, required) — Must match a pipeline that already exists in your Seer dashboard.
capture_logs (bool) — Capture all stdout/stderr and logging output. Logs are synced progressively. Default: False.
metadata (dict) — Custom metadata attached to the run.
tags (list[str]) — Tags attached to the run (e.g. ["etl", "prod"]).

Behavior:

• Automatically tracks start time, end time, and duration
• Captures exceptions and full stack traces
• Sends status updates to SEER API (running → success/failed)
• HTTP 4xx errors are not retried (except 429); 5xx and connection errors use exponential backoff
• If the API is unreachable at start, the job still runs locally and only the final outcome is queued — no forever-running stubs
• Monitoring never raises — Seer outages cannot fail your job

Example:

with seer.monitor(
    "etl_pipeline",
    capture_logs=True,
    metadata={"source": "postgres", "target": "s3", "env": "production"},
    tags=["etl", "prod"],
):
    extract_data()
    transform_data()
    load_data()

seer.heartbeat(job_name, metadata=None, tags=None)

Send a heartbeat signal to SEER

Parameters:

job_name (str, required) — Name of the job sending the heartbeat.
metadata (dict) — Additional context (e.g., progress, status).
tags (list[str]) — Tags attached to the heartbeat.

Example:

import time

for batch in range(100):
    process_batch(batch)

    if batch % 5 == 0:
        seer.heartbeat(
            "batch_processor",
            metadata={"batch": batch, "progress": f"{batch}%"},
            tags=["prod"],
        )

    time.sleep(300)  # 5 minutes

seer.replay(max_attempts=5)

Flush the offline queue

Returns:

A result object with sent, failed, and dead_lettered counts.

Offline queue details:

• Envelopes are stored in ~/.seer/queue (shared with the Seer CLI)
• Each envelope includes: endpoint, payload, created_at, attempts, idempotency_key, and base_url
• Atomic writes (tmp + os.replace) — readers never see partial files
• Cross-process locking via filelock; claim-by-rename (.sending) avoids double-sends
• FIFO eviction at 500 files / 50 MiB (override with env vars)
• After repeated failures, envelopes move to ~/.seer/queue/dead/

Example:

result = seer.replay()
print(result.sent, result.failed, result.dead_lettered)

Error Handling & Offline Mode

SEER automatically captures and reports exceptions with full stack traces. Monitoring never raises — if Seer is down, the final result is queued for replay.

from seerpy import Seer
import pandas as pd

seer = Seer(api_key="YOUR_API_KEY", auto_replay=True)

# Automatic error capture
with seer.monitor("data_processing", capture_logs=True):
    # This error will be automatically captured with full traceback
    data = pd.read_csv("missing_file.csv")
    process_data(data)

# If the API is down, the payload is saved to ~/.seer/queue and
# will be replayed automatically on the next run (auto_replay=True)
# or you can flush manually:
result = seer.replay()
print(f"Sent: {result.sent}, Failed: {result.failed}")

Retry Logic

HTTP 4xx errors are not retried (except 429 Too Many Requests)
HTTP 5xx and connection errors use exponential backoff
Every live POST and queued envelope carries a UUID v4 Idempotency-Key header
Replay uses {key}:register then {key}:complete to avoid duplicates

Best Practices

Do's

Use descriptive job names that match your dashboard (e.g., "daily_sales_etl")
Include relevant metadata for debugging
Keep API keys in environment variables
Use auto_replay=True for scripts, background_replay=True for servers
Use separate jobs for different pipeline stages
Use tags= to group related jobs

Don'ts

Don't hardcode API keys in your scripts
Don't use generic job names like "job1"
Don't wrap individual functions (wrap the whole pipeline)
Don't log sensitive data in metadata or logs
Don't create new Seer instances in loops
Don't use payloads.replay_failed_payloads() — use seer.replay() instead

Environment Variables

Variable	Purpose
SEER_API_KEY	API key (pass into `Seer(api_key=...)`)
SEER_BASE_URL	Override default API host (`https://api.ansrstudio.com/`)
SEER_QUEUE_DIR	Offline queue directory (default `~/.seer/queue`)
SEER_QUEUE_MAX_FILES	Max queued envelopes (default 500)
SEER_QUEUE_MAX_BYTES	Max queue size in bytes (default 50 MiB)

Store your API key securely using environment variables:

export SEER_API_KEY="your_api_key_here"

from seerpy import Seer
import os

seer = Seer(api_key=os.environ["SEER_API_KEY"])

Full Worker Example

import os
from seerpy import Seer

seer = Seer(
    api_key=os.getenv("SEER_API_KEY"),
    auto_replay=True,
    background_replay=True,
    replay_interval=60,
)

def run_job():
    with seer.monitor(
        "example_worker",
        capture_logs=True,
        metadata={"env": "prod"},
        tags=["worker"],
    ):
        print("Starting work...")
        for i in range(3):
            print(f"step {i+1}")
        seer.heartbeat("example_worker", metadata={"progress": "50%"})
        print("Work complete.")

if __name__ == "__main__":
    run_job()
    seer.stop_background_replay()

Additional Resources

Code Examples

Real-world integration examples

Best Practices

Error handling and optimization

Support

Get help and report issues