pingback-py 0.3.1

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

# settings.py
from pingback import Pingback

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
)

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from myproject.settings import pb


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

Register your url:

# urls.py
from django.urls import path
from myapp.views import pingback_handler

urlpatterns = [
    path("api/pingback", pingback_handler),
]

Put your jobs in a jobs.py file inside each app, then register everything from one central AppConfig (listed last in INSTALLED_APPS):

# billing/jobs.py  — just decorate, no register() call needed here
from myproject.settings import pb


@pb.cron("monthly-invoice", "0 9 1 * *")
def monthly_invoice(ctx):
    ...


@pb.task("charge-card")
def charge_card(ctx):
    ...

# myproject/apps.py  — one registration point for the whole project
from django.apps import AppConfig


class MyProjectConfig(AppConfig):
    name = "myproject"

    def ready(self):
        from django.utils.module_loading import autodiscover_modules
        from myproject.settings import pb

        autodiscover_modules(
            "tasks")  # auto-imports tasks.py (when using jobs.py -- autodiscover_modules("jobs") from every installed app 
        pb.register()  # called once after all decorators have run

Adding a new job only requires editing (or creating) the relevant app's tasks.py or jobs.py — no apps.py changes, no extra pb.register() calls.

NB: Where tasks/jobs is a package, be sure to include all tasks/jobs in the package's init.py

Any Framework

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

Registration: flask_handler() and fastapi_handler() automatically register your functions with the platform on startup. For Django, use the autodiscover_modules pattern above to register once after all apps are loaded. pb.register() always sends the current set of functions to the platform.

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

Typed Payloads

Task handlers can accept a typed second parameter for autocomplete, validation, and self-documenting code. Works with dataclasses and Pydantic models:

from dataclasses import dataclass


@dataclass
class EmailPayload:
    to: str
    subject: str
    priority: int = 1


@pb.task("send-email", retries=3)
def send_email(ctx, payload: EmailPayload):
    # payload.to, payload.subject — full autocomplete
    send_mail(payload.to, payload.subject)
    ctx.log("Sent", to=payload.to, priority=payload.priority)

With Pydantic (pip install pydantic):

from pydantic import BaseModel


class OrderPayload(BaseModel):
    order_id: str
    amount: float
    email: str


@pb.task("process-order")
def process_order(ctx, payload: OrderPayload):
    # validated, with defaults and type coercion
    ctx.log("Processing", order_id=payload.order_id)

Unpacked Kwargs

Task handlers can receive payload fields directly as keyword arguments — no need to extract from a payload object:

@pb.task("send-password-reset", retries=3)
def send_password_reset(ctx, otp_code: str, user_email: list[str]):
    message = f"Your OTP is {otp_code}."
    send_mail(message=message, recipient_list=user_email)
    ctx.log("Sent reset email", to=user_email)

Triggered with:

pb.trigger("send-password-reset", {"otp_code": "482910", "user_email": ["[e-mail]"]})

This activates automatically when unpack_payload=True (the default) and the handler has more than one parameter beyond ctx, or a single extra parameter not named payload. The SDK unpacks ctx.payload as keyword arguments into the function. Set unpack_payload=False to disable this and use the raw dict or typed payload styles instead.

All four styles are supported:

Style	Signature	Payload access
No param	def job(ctx)	ctx.payload["key"]
Raw dict	def job(ctx, payload)	payload["key"]
Typed	def job(ctx, payload: MyType)	payload.key
Unpacked kwargs	def job(ctx, field1, field2)	field1, field2 directly

Fan-Out

@pb.cron("send-emails", "*/15 * * * *")
def send_emails(ctx):
    pending = get_pending_emails()
    for email in pending:
        ctx.task("send-email", {"to": email.recipient, "subject": email.subject})
    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:

@dataclass
class Order:
    order_id: str
    amount: float
    email: str


@pb.task("validate-order", retries=2)
def validate_order(ctx, order: Order):
    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_id": order.order_id, "amount": order.amount, "email": order.email})
    return {"valid": True}


@pb.task("charge-payment", retries=3)
def charge_payment(ctx, payload: Order):
    charge = stripe.Charge.create(amount=int(payload.amount * 100))
    ctx.log("Charged", charge_id=charge.id)
    ctx.task("send-confirmation", {"email": payload.email, "order_id": payload.order_id})


@pb.task("send-confirmation", retries=2)
def send_confirmation(ctx, payload):
    send_email(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]"})

# With a delay — run 15 minutes from now
exec_id = pb.trigger("send-email", {"to": "[e-mail]"}, delay="15m")

# Delay as seconds
exec_id = pb.trigger("send-email", {"to": "[e-mail]"}, delay=900)

Supported delay formats: integer (seconds), or a string like "30s", "15m", "2h", "1d", "1d2h30m". Maximum delay: 30 days.

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, unpack_payload=True)

unpack_payload (default True) — when the handler has multiple parameters beyond ctx, or a single extra parameter not named payload, the SDK unpacks ctx.payload as keyword arguments. Set to False to always use the raw dict / typed payload styles.

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. Functions are registered with the platform on startup (flask_handler() and fastapi_handler() do this automatically; for Django or other frameworks, call pb.register())
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 and workflow chains are dispatched independently by the platform