Official Python SDK for SEER - Monitor your data pipelines with simple context managers
pip install seerpy
Requirements: Python 3.7 or higher
# Import SEER monitoring
from seerpy import Seer
# Initialize with your API key
seer = Seer(apiKey='YOUR_API_KEY')
# Your existing imports
import pandas as pd
def run_report():
# Your script logic here
data = pd.read_csv("sales_data.csv")
processed = data.groupby("region").sum()
processed.to_csv("daily_report.csv")
# Wrap your script with SEER monitoring
with seer.monitor("daily-report-job"):
run_report() # Your script logic here
# Output:
# ✓ Connected to SEER monitoring
# ✓ Pipeline "daily-report-job" registered
# → Monitoring active.
# ✓ Monitoring complete.✓ That's it! SEER will automatically track execution time, success/failure status, and send notifications if something goes wrong.
apiKey (str, required) - Your SEER API key from the dashboardfrom seerpy import Seer seer = Seer(apiKey='df_your_api_key_here')
job_name (str, required) - Unique identifier for your pipeline/jobcapture_logs (bool, optional) - Capture all stdout/stderr and logging output. Default: Falsemetadata (dict, optional) - Custom metadata to attach to the run (e.g., config, parameters)with seer.monitor(
"etl-pipeline",
capture_logs=True,
metadata={
"source": "postgres",
"target": "s3",
"env": "production"
}
):
# Your code here
extract_data()
transform_data()
load_data()job_name (str, required) - Name of the job sending the heartbeatmetadata (dict, optional) - Additional context about the heartbeat (e.g., progress, status)import time
for batch in range(100):
process_batch(batch)
# Send heartbeat every 5 minutes
if batch % 5 == 0:
seer.heartbeat(
"batch-processor",
metadata={"batch": batch, "progress": f"{batch}%"}
)
time.sleep(300) # 5 minutesSEER automatically captures and reports exceptions with full stack traces:
from seerpy import Seer
import pandas as pd
seer = Seer(apiKey='YOUR_API_KEY')
# Automatic error capture
with seer.monitor("data-processing"):
# This error will be automatically captured with full traceback
data = pd.read_csv("missing_file.csv")
process_data(data)
# Output on error:
# ✓ Connected to SEER monitoring
# ✓ Pipeline "data-processing" registered
# → Monitoring active.
# FileNotFoundError: [Errno 2] No such file or directory: 'missing_file.csv'
# ✓ Monitoring complete.
# The error is sent to SEER with:
# - Full stack trace
# - Error type and message
# - Timestamp
# - All metadata
# - Captured logs (if enabled)
# Offline mode - if SEER API is unreachable
with seer.monitor("critical-job"):
process_data()
# If the API is down, the payload is saved locally and will be
# replayed when the connection is restoredSEER uses exponential backoff for API calls:
If payloads were saved offline and you want to manually replay them, use this function:
Using the payloads module (recommended):
from seerpy import Seer, payloads
import time
# Your main script
seer = Seer(api_key='your_api_key')
with seer.monitor(job_name='daily-etl'):
# Your code here
time.sleep(2)
# Replay any failed payloads from previous runs
payloads.replay_failed_payloads('your_api_key')Manual implementation:
import os
import json
import requests
def replay_failed_payloads(api_key: str):
temp_dir = os.path.join(os.path.dirname(__file__), "failed_payloads")
if not os.path.exists(temp_dir):
return
for filename in os.listdir(temp_dir):
filepath = os.path.join(temp_dir, filename)
with open(filepath, "r") as f:
payload = json.load(f)
headers = {
"Authorization": api_key,
"Content-Type": "application/json"
}
if "monitoring" in filename:
url = "https://api.seer.ansrstudio.com/monitoring"
elif "heartbeat" in filename:
url = "https://api.seer.ansrstudio.com/heartbeat"
else:
continue
try:
requests.post(url, headers=headers, json=payload)
os.remove(filepath) # Delete on success
except Exception:
continue # Leave the file for next retry
# Usage
replay_failed_payloads("YOUR_API_KEY")Store your API key securely using environment variables:
# Set environment variable export SEER_API_KEY="df_your_api_key_here" # Use in your script from seerpy import Seer import os seer = Seer(apiKey=os.environ['SEER_API_KEY'])
API Rate Limits:
Free Plan:
Pro Plan: