Trigger an Excel Report from a Webhook
Most report jobs run on a schedule — every morning at seven, every Monday. Some reports are better built when something happens: a month is closed in the accounting system, an order batch is approved, a survey closes, a pipeline finishes loading data. Many systems can announce such events by calling a URL you give them — a webhook — with a small JSON payload. A Python endpoint that receives that call can build the Excel report straight away, so it arrives minutes after the event rather than the next morning, and never runs on stale data. Doing it safely takes more than a route that calls the report function: the endpoint must reject calls that did not come from the real sender, cope with the same event being delivered twice, and answer quickly even if the report takes a minute to build. This page builds such an endpoint with FastAPI and tests it without deploying anything. It belongs to Scheduling Python Excel Scripts with Cron.
Prerequisites
pip install fastapi uvicorn httpx pandas openpyxl
httpx is needed by FastAPI's test client. The sending system and its secret are illustrative: most webhook providers — accounting and payment platforms, form tools, GitHub, CI services — sign their requests with HMAC-SHA256 in much the same way, but the header names and the exact string that is signed differ, so check your provider's documentation for those two details.
Step 1 — The report the event triggers
The report itself is an ordinary function that takes the event's data — here, which period was closed — and writes a workbook:
from pathlib import Path
import pandas as pd
OUT = Path("reports")
OUT.mkdir(exist_ok=True)
def build_close_report(period):
ledger = pd.DataFrame({"account": ["4000 Sales", "5000 Cost of sales", "6100 Wages", "6200 Rent"],
"amount": [182400.0, -96350.0, -41200.0, -8500.0]})
path = OUT / f"close_report_{period}.xlsx"
with pd.ExcelWriter(path) as xw:
ledger.to_excel(xw, sheet_name="Ledger", index=False)
pd.DataFrame({"period": [period], "result": [ledger["amount"].sum()]}).to_excel(xw, sheet_name="Summary", index=False)
return path
In a real job this would query the accounting system for the closed period; the point is that it is the same function a scheduled job would call. Keeping it separate from the web endpoint means it can be tested, rerun by hand and triggered by other means. Making it safe to run twice for the same period — deterministic file names, atomic writes — is covered in Make an Excel Report Job Safe to Rerun.
Step 2 — Verify the signature and timestamp
Anyone who learns the URL can POST to it, so the endpoint must check that each request came from the real sender. The sender and receiver share a secret; the sender computes an HMAC-SHA256 of the timestamp and the raw body with that secret and puts it in a header. The receiver recomputes it and compares:
import hashlib
import hmac
import time
SECRET = b"whsec_test_3f9a1c" # in production: from an environment variable or secret store
TOLERANCE = 300 # seconds
def sign(body: bytes, timestamp: int, secret: bytes = SECRET) -> str:
message = str(timestamp).encode() + b"." + body
return "sha256=" + hmac.new(secret, message, hashlib.sha256).hexdigest()
def verify(body: bytes, timestamp: str | None, signature: str | None, now: float | None = None) -> str | None:
if not timestamp or not signature:
return "missing signature headers"
try:
ts = int(timestamp)
except ValueError:
return "bad timestamp"
if abs((now or time.time()) - ts) > TOLERANCE:
return "timestamp outside tolerance"
if not hmac.compare_digest(sign(body, ts), signature):
return "signature mismatch"
return None
Three details matter. The signature is computed over the raw bytes of the body, exactly as received — parsing the JSON and re-serialising it changes whitespace or key order and breaks the comparison. hmac.compare_digest compares in constant time, so an attacker cannot discover the correct signature byte by byte from response timings. And including a timestamp in the signed message, then rejecting requests more than five minutes old, stops someone who captured a genuine request from replaying it later.
Step 3 — The endpoint: verify, deduplicate, accept, work later
from fastapi import BackgroundTasks, FastAPI, Header, Request
from fastapi.responses import JSONResponse
app = FastAPI()
seen_events: set[str] = set() # use a database table in production
built: list[str] = []
def run_report(period: str):
built.append(build_close_report(period).name)
@app.post("/webhooks/close")
async def period_closed(request: Request, background: BackgroundTasks,
x_webhook_timestamp: str | None = Header(None), x_webhook_signature: str | None = Header(None)):
body = await request.body()
problem = verify(body, x_webhook_timestamp, x_webhook_signature)
if problem:
return JSONResponse({"error": problem}, status_code=401)
event = await request.json()
if event["id"] in seen_events:
return JSONResponse({"status": "duplicate", "id": event["id"]}, status_code=200)
seen_events.add(event["id"])
if event.get("type") != "period.closed":
return JSONResponse({"status": "ignored", "type": event.get("type")}, status_code=200)
background.add_task(run_report, event["data"]["period"])
return JSONResponse({"status": "accepted", "id": event["id"]}, status_code=202)
The endpoint reads the raw body first, verifies it, and only then parses the JSON. Each event carries an ID, and webhook senders deliver at least once: if your response is slow or lost, they retry, so the same event can arrive two or three times. Recording IDs and acknowledging repeats without acting on them prevents duplicate reports; in production the set must be a database table (with a unique constraint) so it survives restarts and works across several server processes. Events of other types are acknowledged and ignored, so the sender does not keep retrying them. Finally the report is handed to BackgroundTasks and the endpoint returns 202 Accepted immediately — most senders wait only a few seconds for a response before treating the delivery as failed and retrying it.
Step 4 — Test it end to end
FastAPI's TestClient sends real HTTP requests to the app in-process, background tasks included, so the whole flow can be tested without a server:
import json
from fastapi.testclient import TestClient
client = TestClient(app)
def deliver(event, secret=SECRET, timestamp=None):
body = json.dumps(event).encode()
ts = timestamp or int(time.time())
headers = {"X-Webhook-Timestamp": str(ts), "X-Webhook-Signature": sign(body, ts, secret), "Content-Type": "application/json"}
r = client.post("/webhooks/close", content=body, headers=headers)
return r.status_code, r.json()
event = {"id": "evt_101", "type": "period.closed", "data": {"period": "2026-09"}}
print("genuine: ", deliver(event))
print("redelivered: ", deliver(event))
print("wrong secret: ", deliver({**event, "id": "evt_102"}, secret=b"guess"))
print("replayed old: ", deliver({**event, "id": "evt_103"}, timestamp=int(time.time()) - 3600))
print("other type: ", deliver({"id": "evt_104", "type": "invoice.paid", "data": {}}))
print("reports built:", built)
The genuine event is accepted and the report is built once; the redelivery of the same event is acknowledged as a duplicate and builds nothing; a request signed with the wrong secret and a replay of an hour-old request are both rejected with 401; an event of another type is acknowledged and ignored. The final line shows a single report file — exactly one, despite five deliveries. These assertions belong in your test suite, with a fixed secret for tests and the real one read from the environment in production; Store Credentials for Excel Report Scripts Safely covers the latter.
Running it for real
Run the app with uvicorn webhook_app:app --host 0.0.0.0 --port 8000 behind HTTPS — webhook senders require it, and the signature protects authenticity but not confidentiality. BackgroundTasks runs the report in the same process after the response; that is fine for reports that take seconds to a minute. For longer reports, or if losing an in-progress report on a restart is unacceptable, have the endpoint write the event to a queue or database table and let a separate worker build reports, as described in Generate Large Excel Exports in a Background Job. Log every delivery with its event ID and outcome, and keep a scheduled fallback — a nightly job that builds any closed period without a report — so a webhook the sender never delivered does not mean a report nobody receives.
Common pitfalls
- Verifying parsed JSON. Sign and verify the raw request body.
- Plain
==on signatures. Usehmac.compare_digest. - No timestamp check. Captured requests can be replayed later.
- Building the report before responding. The sender times out and retries; return 202 and work in the background.
- Assuming one delivery per event. Record event IDs and ignore repeats.
Conclusion
To build an Excel report when an event happens, expose a webhook endpoint that verifies an HMAC-SHA256 signature over the timestamp and raw body with hmac.compare_digest, rejects stale timestamps, records event IDs so redeliveries are ignored, and returns 202 Accepted at once while the report is built in the background. Test the genuine, duplicate, forged and replayed cases with FastAPI's test client, and keep a scheduled fallback for events that never arrive.
Frequently asked questions
How do I verify a webhook signature in Python? Recompute an HMAC-SHA256 over the exact raw request body (and timestamp, if the provider signs it) with the shared secret, and compare it with the header value using hmac.compare_digest.
Why does my webhook get delivered twice? Senders retry when a response is slow or lost, so delivery is at least once; record each event ID and acknowledge repeats without processing them again.
Should a webhook endpoint build the report before responding? No. Return 202 Accepted quickly and build the report in a background task or worker, because senders time out after a few seconds and retry.
How do I test a FastAPI webhook without deploying it? Use fastapi.testclient.TestClient to post signed requests to the app in-process; background tasks run as part of the request.
Related
- Up one level: Scheduling Python Excel Scripts with Cron — running reports automatically.
- Run a Report When an Excel File Changes in OneDrive — Microsoft Graph change notifications.
- Watch a Folder and Process New Excel Files with Python — reacting to files instead of events.
- Stream an Excel File from a FastAPI Endpoint — FastAPI for downloads.
More on scheduling & orchestration
- How-toDownload New Excel Files from an SFTP Server
- How-toRead Excel Attachments into a Database Nightly
- How-toRefresh an Excel Report from a Database on a Schedule
- How-toRun a Report When an Excel File Changes in OneDrive
- How-toSend Report Emails at Each Recipient's Local Time
- How-toDetect a Scheduled Report That Did Not Run