pingback-py 0.1.0

Python SDK for Pingback — reliable cron jobs and background tasks

pingback-py

Python SDK for Pingback — reliable cron jobs and background tasks.

Installation

pip install pingback-py

Quick Start

import os
from pingback import Pingback

pb = Pingback(
    api_key=os.environ["PINGBACK_API_KEY"],
    cron_secret=os.environ["PINGBACK_CRON_SECRET"],
)

@pb.cron("cleanup", "0 3 * * *", retries=2, timeout="60s")
def cleanup(ctx):
    removed = remove_expired_sessions()
    ctx.log("Removed sessions", count=removed)
    return {"removed": removed}

@pb.task("send-email", retries=3, timeout="15s")
def send_email(ctx):
    to = ctx.payload["to"]
    deliver_email(to)
    ctx.log("Sent email", to=to)
    return {"sent": to}

Framework Integration

Flask

from flask import Flask

app = Flask(__name__)
app.route("/api/pingback", methods=["POST"])(pb.flask_handler())

FastAPI

from fastapi import FastAPI

app = FastAPI()
app.post("/api/pingback")(pb.fastapi_handler())

Django

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def pingback_handler(request):
    result = pb.handle(request.body, dict(request.headers))
    status = result.pop("_status", 200)
    return JsonResponse(result, status=status)

Any Framework

result = pb.handle(body=request_body_bytes, headers=request_headers_dict)

Defining Functions

Cron Jobs

@pb.cron("daily-report", "0 9 * * *", retries=3, timeout="60s")
def daily_report(ctx):
    report = generate_report()
    ctx.log("Report generated", rows=report.row_count)
    return report

Background Tasks

@pb.task("process-upload", retries=2, timeout="5m")
def process_upload(ctx):
    file_id = ctx.payload["file_id"]
    result = process_file(file_id)
    ctx.log("Processed file", file_id=file_id)
    return result

Fan-Out

@pb.cron("send-emails", "*/15 * * * *")
def send_emails(ctx):
    pending = get_pending_emails()
    for email in pending:
        ctx.task("send-email", {"id": email.id})
    ctx.log("Dispatched emails", count=len(pending))
    return {"dispatched": len(pending)}

Workflows (Task Chaining)

Tasks can call ctx.task() to chain into multi-step workflows with branching:

@pb.task("validate-order", retries=2)
def validate_order(ctx):
    order = ctx.payload
    ctx.log("Validating", order_id=order["order_id"])

    if order["amount"] <= 0:
        ctx.task("notify-failure", {"order_id": order["order_id"], "reason": "Invalid amount"})
        return {"valid": False}

    ctx.task("charge-payment", order)
    return {"valid": True}

@pb.task("charge-payment", retries=3)
def charge_payment(ctx):
    charge = stripe.Charge.create(amount=ctx.payload["amount"] * 100)
    ctx.log("Charged", charge_id=charge.id)
    ctx.task("send-confirmation", ctx.payload)

@pb.task("send-confirmation", retries=2)
def send_confirmation(ctx):
    send_email(ctx.payload["email"], "Order confirmed")
    ctx.log("Confirmation sent")

Each step runs as its own execution with independent retries and logging. The workflow graph in your dashboard visualizes the full chain.

Programmatic Triggering

exec_id = pb.trigger("send-email", {"to": "[e-mail]"})

Structured Logging

ctx.log("message")                         # info
ctx.log("message", key="value")            # info with metadata
ctx.warn("slow query", ms=2500)            # warning
ctx.error("failed", code="E001")           # error
ctx.debug("cache stats", hits=847)         # debug

Configuration

pb = Pingback(
    api_key="pb_live_...",
    cron_secret="...",
    platform_url="https://api.pingback.lol",  # default
    base_url="https://myapp.com",              # your app's public URL
)

Function Options

@pb.cron("job", "* * * * *", retries=3, timeout="30s", concurrency=5)
@pb.task("job", retries=3, timeout="30s", concurrency=5)

Environment Variables

PINGBACK_API_KEY=pb_live_...        # From your Pingback project settings
PINGBACK_CRON_SECRET=...            # From your Pingback project settings

How It Works

1. Define cron jobs and tasks with @pb.cron() and @pb.task() decorators
2. Mount the handler using your framework's routing
3. On the first request, the SDK registers your functions with the Pingback platform
4. The platform sends signed HTTP requests to your handler when jobs are due
5. The handler verifies the HMAC signature, executes the function, and returns results
6. Fan-out tasks are dispatched independently by the platform