Countdown Timer API with Python
A small client built on requests, with retry handling that distinguishes the two kinds of 429, and webhook handlers for FastAPI, Django, and Flask. Each framework gets the raw request body a slightly different way, and getting that wrong is the single most common reason signature verification fails — so all three are shown explicitly.
The client
A session for connection reuse, a typed error carrying the request_id, and a timeout. That last one matters: requests has no default timeout, so a call without one can hang indefinitely.
countdownshare.py
# countdownshare.py — requires: pip install requests
import os
import time
from typing import Any
import requests
BASE = "https://countdownshare.com/api/v1"
RETRYABLE = {"rate_limited", "internal_error"}
class CountdownError(Exception):
def __init__(self, code: str, message: str, status: int, request_id: str,
retry_after: int | None = None):
super().__init__(f"{code}: {message} ({request_id})")
self.code = code
self.status = status
self.request_id = request_id
self.retry_after = retry_after
class Countdown:
def __init__(self, api_key: str | None = None):
self.session = requests.Session()
key = api_key or os.environ["COUNTDOWNSHARE_API_KEY"]
self.session.headers.update({
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
})
def request(self, method: str, path: str, **kwargs: Any) -> dict:
response = self.session.request(method, BASE + path, timeout=10, **kwargs)
payload = response.json()
if not response.ok:
error = payload["error"]
raise CountdownError(
error["code"],
error["message"],
response.status_code,
payload["request_id"],
int(response.headers.get("Retry-After", 0)) or None,
)
return payload["data"]Retries and creation
Retry, create, read
def _with_retry(self, method: str, path: str, attempts: int = 4, **kwargs):
"""Retry only what will actually clear.
quota_exhausted is a 429 too, but the monthly allowance does
not refill until the billing cycle resets — branch on the
code, never on the status alone.
"""
for attempt in range(attempts):
try:
return self.request(method, path, **kwargs)
except CountdownError as error:
last = attempt == attempts - 1
if error.code not in RETRYABLE or last:
raise
time.sleep(error.retry_after or min(2 ** attempt * 0.5, 8))
def create_timer(self, body: dict, idempotency_key: str) -> dict:
return self._with_retry(
"POST", "/timers",
headers={"Idempotency-Key": idempotency_key},
json=body,
)
def status(self, timer_id: str) -> dict:
return self.request("GET", f"/timers/{timer_id}/status")rate_limited and internal_error are retried. A validation_error fails identically every time, and quota_exhausted will not clear until the billing cycle resets — see error codes.Creating a timer
The Idempotency-Key is derived from the customer rather than generated at the call site, which is what makes a retried signup safe.
Per-customer trial deadline
client = Countdown()
# The key is derived from the customer, so a retried signup
# returns the original timer instead of granting a second,
# longer trial.
timer = client.create_timer(
{
"name": f"Trial — {customer.email}",
"type": "personalized",
"duration_seconds": 14 * 24 * 60 * 60,
"external_user_id": str(customer.id),
"publish": True,
},
idempotency_key=f"trial_{customer.id}",
)
customer.countdown_timer_id = timer["id"]
customer.save()
# Later, anywhere:
if client.status(customer.countdown_timer_id)["ended"]:
downgrade(customer)Store the returned id against your own record. It is what every later call takes — status, updates, webhook rules, metrics.
Verifying a webhook
The signature is an HMAC-SHA256 over "{timestamp}.{raw_body}". One helper, shared by every framework.
verify.py
# verify.py — shared by every framework below.
import hashlib
import hmac
import time
def verify(raw_body: bytes, headers, secret: str) -> bool:
timestamp = headers.get("X-CountdownShare-Timestamp")
signature = headers.get("X-CountdownShare-Signature")
if not timestamp or not signature:
return False
# Reject anything older than five minutes: a captured
# signature is otherwise replayable forever.
if abs(time.time() - int(timestamp)) > 300:
return False
expected = hmac.new(
secret.encode(),
f"{timestamp}.".encode() + raw_body,
hashlib.sha256,
).hexdigest()
received = signature.removeprefix("v1=")
return hmac.compare_digest(expected, received) # constant timehmac.compare_digest rather than ==. A normal string comparison short-circuits on the first differing character, which leaks how much of a guessed signature was correct.Getting the raw body in each framework
This is where Python integrations go wrong. Every framework offers a convenient parsed body, and every one of them destroys what you need to verify against.
FastAPI
FastAPI
from fastapi import BackgroundTasks, FastAPI, Request, Response
app = FastAPI()
@app.post("/webhooks/countdownshare")
async def countdown_webhook(
request: Request,
background: BackgroundTasks,
):
# await request.body() gives the RAW bytes. Do not use
# request.json() before verifying — the bytes are then gone.
raw = await request.body()
if not verify(raw, request.headers, os.environ["COUNTDOWNSHARE_WEBHOOK_SECRET"]):
return Response(status_code=401)
event = json.loads(raw)
# Return inside 10 seconds; do the work afterwards.
background.add_task(handle_countdown_event, event)
return Response(status_code=200)Django and Flask
Django and Flask
# Django — request.body is the raw bytes. request.POST is not.
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
@csrf_exempt # an external caller has no CSRF token
@require_POST
def countdown_webhook(request):
if not verify(request.body, request.headers, settings.COUNTDOWN_WEBHOOK_SECRET):
return HttpResponse(status=401)
event = json.loads(request.body)
handle_countdown_event.delay(event) # Celery
return HttpResponse(status=200)
# Flask — request.get_data(), NOT request.json
@app.post("/webhooks/countdownshare")
def countdown_webhook():
raw = request.get_data()
if not verify(raw, request.headers, os.environ["COUNTDOWNSHARE_WEBHOOK_SECRET"]):
return "", 401
handle_countdown_event.delay(json.loads(raw))
return "", 200await request.body(), request.body, and request.get_data() give you bytes. request.json(), request.POST, and request.json give you a parsed object that cannot be turned back into the original bytes. Verify first, parse second.The Django example also needs @csrf_exempt. An external caller has no CSRF token, so without it Django rejects the request before your view runs — and the HMAC signature is doing the authentication job anyway.
Common questions
Is there a Python SDK on PyPI?
Not currently. Every example in the documentation is plain HTTP, and the client on this page is short enough to paste into a project. The OpenAPI 3.1 spec is public if you would rather generate a typed client with openapi-python-client or similar.
Why does my Django webhook return 403?
CSRF. Django rejects an unauthenticated POST from an external origin before your view runs. Decorate the view with @csrf_exempt — the HMAC signature is what authenticates the request, and it is stronger than a CSRF token for this purpose.
Why does signature verification fail in Flask?
Almost always because request.json was read first. Use request.get_data() to get the raw bytes. Once Flask parses the body, re-serialising it produces different whitespace and key ordering, so the HMAC can never match.
Should I use async or sync?
Either. The client here is sync because requests is the most common choice and timer calls are infrequent — you create a timer when a record is created, not on every request. If you are already on httpx and async, the same structure ports directly.
Next steps
Run it against Sandbox
Free with any account, no plan needed. Set COUNTDOWNSHARE_API_KEY and the client works as written.