Python

Set up Pingback in your Python app with the pingback-py SDK.

Installation

pip install pingback-py

Setup

Create a Pingback client and define your functions with decorators:

import os
from pingback import Pingback

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

Defining Functions

Use @pb.cron() and @pb.task() decorators to register functions. The return value becomes the execution result:

@pb.cron("send-emails", "*/15 * * * *", retries=3, timeout="60s")
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)}

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

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

Use the raw handle() method with any framework:

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

Structured Logging

Log with keyword arguments for structured metadata:

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

Programmatic Triggering

Trigger tasks from anywhere in your code:

exec_id = pb.trigger("send-email", {"to": "user@example.com"})

Fan-Out

Dispatch background tasks from within a cron handler using ctx.task(). Each task runs independently with its own retries and timeout:

@pb.cron("process-orders", "0 * * * *", retries=3, timeout="120s")
def process_orders(ctx):
    orders = get_unprocessed_orders()
    for order in orders:
        ctx.task("fulfill-order", {"order_id": order.id})
    ctx.log("Dispatched orders", count=len(orders))
    return {"dispatched": len(orders)}

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

Option	Default	Description
`retries`	0	Retry up to n times on failure.
`timeout`	"30s"	Execution timeout (e.g. "30s", "5m").
`concurrency`	1	Max concurrent runs.

Environment Variables

PINGBACK_API_KEY=pb_live_your_api_key_here
PINGBACK_CRON_SECRET=your_cron_secret_here

How It Works

1. Define functions with @pb.cron() and @pb.task() decorators.

2. Mount the handler using your framework's routing (Flask, FastAPI, Django, or raw).

3. On the first request, the SDK registers your functions with the Pingback platform.

4. The platform sends HMAC-signed HTTP requests to your handler on schedule.

5. The handler verifies the signature, executes the function, and returns results with logs.