PDF Tamper Detection in Python: httpx Client, FastAPI and Celery Worker, pytest and Async Batches

This article is a snapshot – content was accurate as of May 2026 (code examples tested against the API as of October 2026). The product evolves actively; specific counts, examples, and detection rules may have changed since publication – see the changelog for the current state.
Your application takes in PDFs from people you cannot vouch for: loan applications, payslips, invoices, insurance claims. Some of them were edited after the issuer produced them, and you have no original to compare against. This walkthrough builds the Python side of that check against the HTPBE? API: one small module with a sync and an async client, a typed result, verdict handling, an error table, a pytest file that runs on test keys, an asyncio batch, and a FastAPI endpoint with a Celery worker for production traffic.
It is deliberately narrow. If you want the concepts first, read How to Detect PDF Tampering Programmatically. If you want a requests.Session client class with Django, Flask and S3 backfill, that is the Python integration guide. If you want to inspect PDF internals yourself with pypdf, see Detect a Tampered PDF in Python Without the Original. This post is the compact, testable version: httpx, frozen dataclasses, retries that cannot double-charge, and tests you can run in CI without a single real PDF.
TL;DR
- Two calls:
POST /v1/analyzereturns{ "id": "..." }with201, thenGET /v1/result/{id}returns the verdict. Analysis is synchronous, so there is nothing to poll. statusisintact,modifiedorinconclusive. None of them is a reason to auto-reject a person; every result carries ausage_cautionobject that says so.- Send an
Idempotency-Keyon everyPOST, reuse it on retries, and a timeout can never spend a credit twice. - Retry
409,429,500and gateway502/503/504only. Everything else is a permanent answer. - Test keys return fixed results for fixed URLs, so the whole decision layer is testable without spending credits.
Prerequisites
pip install httpx pytestPython 3.10 or newer (the code uses X | None annotations). Sign up, then open the API keys page in the dashboard. Create a live key there and copy it when it is shown (htpbe_live_...); your test key (htpbe_test_...) is already on the same page, one per account.
export HTPBE_API_KEY="htpbe_live_..."
export HTPBE_TEST_KEY="htpbe_test_..."The API downloads the file itself, so url must be a public http(s) address that serves a PDF of at most 10 MB without authentication. For private storage, mint a short-lived presigned URL per check (S3, GCS and R2 all support it) and pass that. Files on localhost, behind a VPN or behind a login fail with download_failed.
Step 1: The client module
Put everything below in one file, htpbe_client.py. It has no dependency except httpx, and it is code you own, not a package to pin.
First the types. HtpbeError carries the machine-readable code from the API; CheckResult keeps only the fields a routing decision needs and fails loudly if the API ever returns a status your code does not know.
from __future__ import annotations
import asyncio
import os
import random
import time
import uuid
from dataclasses import dataclass
from typing import Any, Literal
import httpx
BASE_URL = "https://api.htpbe.tech/v1"
# The API allows 30 s to download your file plus 20 s to analyse it.
TIMEOUT = httpx.Timeout(60.0, connect=5.0)
RETRYABLE_STATUS = {409, 429, 500, 502, 503, 504}
MAX_RETRIES = 4
Status = Literal["intact", "modified", "inconclusive"]
Action = Literal["proceed", "review", "request_from_issuer", "unverified"]
_STATUSES = ("intact", "modified", "inconclusive")
class HtpbeError(Exception):
def __init__(
self, status_code: int, code: str, message: str, retry_after: float | None = None
) -> None:
super().__init__(f"{status_code} {code}: {message}")
self.status_code = status_code
self.code = code
self.retry_after = retry_after
@property
def is_retryable(self) -> bool:
return self.status_code in RETRYABLE_STATUS
@dataclass(frozen=True)
class CheckResult:
id: str
status: Status
status_reason: str | None
confidence: str | None
markers: tuple[str, ...]
creator: str | None
producer: str | None
recommended_action: str
signature_removed: bool
modifications_after_signature: bool
@classmethod
def from_json(cls, data: dict[str, Any]) -> CheckResult:
status = data["status"]
if status not in _STATUSES:
raise ValueError(f"Unexpected status: {status!r}")
return cls(
id=data["id"],
status=status,
status_reason=data.get("status_reason"),
confidence=data.get("modification_confidence"),
markers=tuple(data["modification_markers"]),
creator=data.get("creator"),
producer=data.get("producer"),
recommended_action=data["usage_caution"]["recommended_action"],
signature_removed=data["signature_removed"],
modifications_after_signature=data["modifications_after_signature"],
)A few choices worth defending:
- One timeout, set on purpose. The API allows 30 seconds to download your file and 20 seconds to analyse it, and typical documents finish in 2–5 seconds. A 60-second read timeout covers the worst case without hanging a worker indefinitely; the 5-second connect timeout fails fast when the network is down.
markersis a tuple of ids.modification_markersholds stableHTPBE_*identifiers, strongest first. Branch on the id; render the human label from the dictionary on the how it works page.recommended_actionis stored as the API sent it. It lives inusage_cautionand isno_action,route_to_human_revieworrequest_from_issuer, andsafe_for_automated_adverse_decisionis alwaysfalse. Thedecide()function below is your own policy layer on top of it, and it adds the one thing the API cannot know: whether the document claims an institutional origin.
Now the transport: auth, error parsing, retries, and the two-call flow.
def _auth_headers(api_key: str | None) -> dict[str, str]:
return {"Authorization": f"Bearer {api_key or os.environ['HTPBE_API_KEY']}"}
def make_client(api_key: str | None = None) -> httpx.Client:
return httpx.Client(base_url=BASE_URL, headers=_auth_headers(api_key), timeout=TIMEOUT)
def make_async_client(api_key: str | None = None) -> httpx.AsyncClient:
return httpx.AsyncClient(base_url=BASE_URL, headers=_auth_headers(api_key), timeout=TIMEOUT)
def _error_from(response: httpx.Response) -> HtpbeError:
try:
body = response.json()
code, message = body["code"], body["error"]
except (ValueError, KeyError, TypeError):
code, message = "unparseable_error", response.text[:200]
retry_after = response.headers.get("Retry-After", "")
return HtpbeError(
response.status_code,
code,
message,
float(retry_after) if retry_after.isdigit() else None,
)
def _delay(attempt: int, retry_after: float | None) -> float:
if retry_after is not None:
return retry_after
return min(2**attempt, 30) + random.random()
def _send(client: httpx.Client, method: str, path: str, **kwargs: Any) -> httpx.Response:
attempt = 0
while True:
try:
response = client.request(method, path, **kwargs)
except httpx.TransportError:
if attempt == MAX_RETRIES:
raise
time.sleep(_delay(attempt, None))
else:
if response.is_success:
return response
error = _error_from(response)
if not error.is_retryable or attempt == MAX_RETRIES:
raise error
time.sleep(_delay(attempt, error.retry_after))
attempt += 1
def check_pdf(
client: httpx.Client, pdf_url: str, *, idempotency_key: str | None = None
) -> CheckResult:
submitted = _send(
client,
"POST",
"/analyze",
json={"url": pdf_url},
headers={"Idempotency-Key": idempotency_key or str(uuid.uuid4())},
)
check_id = submitted.json()["id"]
return CheckResult.from_json(_send(client, "GET", f"/result/{check_id}").json())The retry loop is where most integrations go wrong, so it is worth reading slowly:
- It retries
409(a request with the same key is still running),429(the server is at analysis capacity)500and gateway502/503/504, plus network-level failures. Every other4xxis a permanent answer about your request, and a second attempt gets the same one. - When the response has a
Retry-Afterheader, that value wins over the exponential backoff. - The same
Idempotency-Keyis sent on every attempt of onecheck_pdfcall. If a timeout leaves you unsure whether the server processed the request, the retry returns the original checkidinstead of charging again. A different key would start a second, separately charged analysis, and the same key sent with a differenturlis rejected with422.
Step 2: Turn a result into a decision
def decide(result: CheckResult, *, claims_institutional_origin: bool) -> Action:
if result.status == "modified":
return "review"
if result.status == "intact":
return "proceed"
return "request_from_issuer" if claims_institutional_origin else "unverified"There is no "reject" on purpose. A structural verdict describes the file, not the person who sent it:
intact: no structural evidence of modification, and the file did not come from consumer software, an online editor or a scan.proceedmeans only that this check found nothing: it is not proof that the content is genuine, so keep your own downstream checks.modified: evidence of post-creation modification. Route to a human reviewer.confidencetells you how firmly:certainmeans conclusive evidence (the finding cannot be a false positive), so hold the document and ask the issuer for a fresh copy;highmeans a closer manual look.inconclusive: the integrity check does not apply, which is not a failure and not a pass.status_reasonsays why:consumer_software_origin,online_editor_origin,scanned_document,unverifiable_metadata,filled_form_originorfill_sign_origin. For a CV or a self-written letter this is normal. For a file that claims to come from a bank, an employer or a tax office, it means you should ask for the original from that institution, which is exactly whatrequest_from_issuerencodes. Handle every reason the same way; do not branch on the exact string.
Usage:
from htpbe_client import check_pdf, decide, make_client
with make_client() as client:
result = check_pdf(client, "https://storage.example.com/uploads/payslip-0412.pdf")
print(result.status, result.markers, result.recommended_action)
print(decide(result, claims_institutional_origin=True))Step 3: Errors you will actually see
Every error body has the same shape: error (message), code (stable, branch on this) and optionally details.
| Status | code | What it means | What to do |
|---|---|---|---|
| 400 | invalid_request | Malformed JSON, missing url, or a bad check id on GET | Fix the request. |
| 400 | invalid_url_format | Not an http(s) URL | Fix the URL. |
| 400 | download_failed | The API could not fetch the file (404, 403, login wall, 30 s timeout) | Check that the URL is public and the link has not expired. |
| 401 | missing_api_key, invalid_api_key | No key, or an unknown one | Check the environment variable; test and live keys are not interchangeable. |
| 402 | payment_required | No credits and no active quota | Stop the run, add credits or upgrade. Retrying cannot help. |
| 403 | inactive_client, account_restricted | The key or account is disabled | Contact support. |
| 403 | test_url_required | A test key was used with a real URL | Use a test URL, or switch to the live key. |
| 404 | not_found | Unknown check id, or it belongs to another client | Use the key that created the check. |
| 409 | idempotency_key_in_progress | The same key is still being processed | Wait and retry with the same key. |
| 413 | file_too_large | Over 10 MB | Compress or split the file. |
| 422 | invalid_pdf | Not a PDF, corrupt, password-protected, or analysis hit its limits | Do not retry. Ask for a clean copy. |
| 422 | idempotency_key_fingerprint_mismatch | A key was reused for a different URL | One key per document. |
| 429 | server_at_capacity | The server’s concurrent-analysis limit is reached; this is server-wide, not a per-key quota | Wait for Retry-After, then retry. |
| 500 | internal_error | Transient server failure | Retry; contact support if it persists. |
Two of these deserve a line in your runbook. 402 should page a human, because every document queued behind it will fail the same way. And invalid_pdf on a password-protected file is a result, not an outage: tell the uploader to remove the password and resubmit.
Step 4: An async batch with bounded concurrency
For a backfill or a nightly sweep, asyncio plus a semaphore is enough. The coroutine below reuses the same helpers, so the retry and idempotency behaviour is identical to the sync path.
async def _send_async(
client: httpx.AsyncClient, method: str, path: str, **kwargs: Any
) -> httpx.Response:
attempt = 0
while True:
try:
response = await client.request(method, path, **kwargs)
except httpx.TransportError:
if attempt == MAX_RETRIES:
raise
await asyncio.sleep(_delay(attempt, None))
else:
if response.is_success:
return response
error = _error_from(response)
if not error.is_retryable or attempt == MAX_RETRIES:
raise error
await asyncio.sleep(_delay(attempt, error.retry_after))
attempt += 1
async def check_pdf_async(
client: httpx.AsyncClient, pdf_url: str, *, idempotency_key: str | None = None
) -> CheckResult:
submitted = await _send_async(
client,
"POST",
"/analyze",
json={"url": pdf_url},
headers={"Idempotency-Key": idempotency_key or str(uuid.uuid4())},
)
check_id = submitted.json()["id"]
fetched = await _send_async(client, "GET", f"/result/{check_id}")
return CheckResult.from_json(fetched.json())
@dataclass(frozen=True)
class BatchItem:
url: str
result: CheckResult | None = None
error: HtpbeError | httpx.HTTPError | None = None
async def check_many(
urls: list[str], *, concurrency: int = 4, api_key: str | None = None
) -> list[BatchItem]:
semaphore = asyncio.Semaphore(concurrency)
async with make_async_client(api_key) as client:
async def check_one(url: str) -> BatchItem:
async with semaphore:
try:
return BatchItem(url, result=await check_pdf_async(client, url))
except (HtpbeError, httpx.HTTPError) as exc:
return BatchItem(url, error=exc)
return list(await asyncio.gather(*(check_one(u) for u in urls)))
if __name__ == "__main__":
urls = [
"https://storage.example.com/batch/statement-jan.pdf",
"https://storage.example.com/batch/statement-feb.pdf",
"https://storage.example.com/batch/statement-mar.pdf",
]
for item in asyncio.run(check_many(urls)):
if item.result:
print(item.url, item.result.status, item.result.markers)
else:
print(item.url, "ERROR", item.error)Notes for real runs:
- Start with
concurrency=4and measure. The429limit is server-wide, so a burst from your batch also competes with everyone else’s. The helpers already back off onRetry-After; a small semaphore just keeps you from spending your retries on it. - Failures come back as data. One bad file produces a
BatchItemwitherrorset, and the rest of the batch continues. Decide per code what to do afterwards:invalid_pdfgoes to a ‘re-upload’ list,download_failedto a ‘link expired’ list. - Watch for account-level errors.
check_manydoes not cancel in-flight work, so scan the returned items: if several carry402,401or403, stop feeding further batches and fix the account first. - Persist the check id. A result stays retrievable by id for the client that created it, so store
result.idnext to your document row and re-read it later withGET /v1/result/{id}rather than paying for a second analysis. - This is in-process concurrency. If the work must survive a restart, put it behind a durable queue; Batch PDF Verification at Scale shows that pattern (with BullMQ) and the same idempotency rules.
Step 5: Test everything with pytest and test keys
A test key (htpbe_test_...) never downloads a file and never touches your quota. It accepts only URLs under https://api.htpbe.tech/v1/test/ and returns a fixed result for each, which makes verdict handling testable in CI. clean.pdf is intact, modified-high.pdf and signature-removed.pdf are modified, inconclusive.pdf is inconclusive, and trigger-402.pdf exercises the out-of-credits path, which a test key cannot reach otherwise. The full list is in the testing guide.
import os
import httpx
import pytest
import htpbe_client
from htpbe_client import CheckResult, HtpbeError, check_pdf, decide, make_client
TEST_BASE = "https://api.htpbe.tech/v1/test"
requires_test_key = pytest.mark.skipif(
"HTPBE_TEST_KEY" not in os.environ, reason="HTPBE_TEST_KEY is not set"
)
@pytest.fixture(scope="module")
def client():
key = os.environ["HTPBE_TEST_KEY"]
assert key.startswith("htpbe_test_"), "refusing to run tests with a live key"
with make_client(key) as c:
yield c
@requires_test_key
@pytest.mark.parametrize(
("filename", "status", "action"),
[
("clean.pdf", "intact", "no_action"),
("modified-high.pdf", "modified", "route_to_human_review"),
("inconclusive.pdf", "inconclusive", "request_from_issuer"),
],
)
def test_verdicts(client, filename, status, action):
result = check_pdf(client, f"{TEST_BASE}/{filename}")
assert result.status == status
assert result.recommended_action == action
@requires_test_key
def test_inconclusive_carries_a_reason(client):
result = check_pdf(client, f"{TEST_BASE}/inconclusive-online-editor.pdf")
assert result.status_reason == "online_editor_origin"
assert result.producer == "iLovePDF"
@requires_test_key
def test_removed_signature_is_reported(client):
result = check_pdf(client, f"{TEST_BASE}/signature-removed.pdf")
assert result.status == "modified"
assert result.signature_removed
assert result.markers
@requires_test_key
def test_out_of_credits_is_not_retried(client):
with pytest.raises(HtpbeError) as excinfo:
check_pdf(client, f"{TEST_BASE}/trigger-402.pdf")
assert (excinfo.value.status_code, excinfo.value.code) == (402, "payment_required")
assert not excinfo.value.is_retryable
@requires_test_key
def test_real_url_is_rejected_for_test_keys(client):
with pytest.raises(HtpbeError) as excinfo:
check_pdf(client, "https://example.com/invoice.pdf")
assert excinfo.value.code == "test_url_required"
def make_result(status: str) -> CheckResult:
return CheckResult(
id="00000000-0000-4000-8000-000000000001",
status=status,
status_reason=None,
confidence=None,
markers=(),
creator=None,
producer=None,
recommended_action="no_action",
signature_removed=False,
modifications_after_signature=False,
)
@pytest.mark.parametrize(
("status", "institutional", "expected"),
[
("intact", True, "proceed"),
("modified", False, "review"),
("inconclusive", True, "request_from_issuer"),
("inconclusive", False, "unverified"),
],
)
def test_decide(status, institutional, expected):
assert decide(make_result(status), claims_institutional_origin=institutional) == expected
def test_retry_reuses_the_idempotency_key_and_honours_retry_after(monkeypatch):
sleeps: list[float] = []
monkeypatch.setattr(htpbe_client.time, "sleep", sleeps.append)
keys: list[str] = []
result_body = {
"id": "00000000-0000-4000-8000-000000000001",
"status": "intact",
"modification_markers": [],
"usage_caution": {"recommended_action": "no_action"},
"signature_removed": False,
"modifications_after_signature": False,
}
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST":
keys.append(request.headers["Idempotency-Key"])
if len(keys) == 1:
return httpx.Response(
429,
headers={"Retry-After": "5"},
json={"error": "busy", "code": "server_at_capacity"},
)
return httpx.Response(201, json={"id": result_body["id"]})
return httpx.Response(200, json=result_body)
transport = httpx.MockTransport(handler)
with httpx.Client(base_url=htpbe_client.BASE_URL, transport=transport) as client:
result = check_pdf(client, "https://example.com/a.pdf")
assert result.status == "intact"
assert len(keys) == 2 and keys[0] == keys[1]
assert sleeps == [5.0]Run it with pytest -q. Without HTPBE_TEST_KEY the live tests skip and the offline tests still run: decide() is checked across every verdict, and one test uses httpx.MockTransport to prove that a 429 with Retry-After: 5 sleeps for exactly five seconds and retries with the same Idempotency-Key. The fixture refuses to run if the key is not a test key, so a misconfigured CI job cannot spend live credits. Do not assert on check_date: it is always null for synthetic results.
Step 6: Run It Behind FastAPI and Celery
Inline calls suit a form with a spinner. Once uploads arrive faster than people click, the check belongs in a worker, and the Python pair that fits is FastAPI for the endpoint and Celery for the retries. There is one trap here that the earlier steps do not show: an Idempotency-Key is bound to the exact url it was first sent with. A task that mints a fresh presigned URL on every retry and reuses upload-{upload_id} as the key gets 422 idempotency_key_fingerprint_mismatch on the second attempt. Mint the URL once, in the endpoint, and pass it to the task together with the key.
# app.py
from typing import Literal
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tasks import verify_upload
from storage import presign_get, uploads # your code
app = FastAPI()
State = Literal["pending", "proceed", "review", "request_from_issuer", "unverified", "failed"]
class VerificationOut(BaseModel):
upload_id: str
state: State
check_id: str | None = None
markers: list[str] = []
error_code: str | None = None
@app.post("/uploads/{upload_id}/verify", status_code=202, response_model=VerificationOut)
def request_verification(upload_id: str) -> VerificationOut:
upload = uploads.get(upload_id)
if upload is None:
raise HTTPException(status_code=404)
# One URL for the life of the task: it must outlive the whole retry window
# (the eight retries below add up to about an hour, so two hours here).
url = presign_get(upload.key, expires_in=7200)
verify_upload.delay(upload_id, url)
return VerificationOut(upload_id=upload_id, state="pending")
@app.get("/uploads/{upload_id}/verification", response_model=VerificationOut)
def read_verification(upload_id: str) -> VerificationOut:
record = uploads.verification(upload_id)
if record is None:
raise HTTPException(status_code=404)
return VerificationOut(**record)# tasks.py
from celery import Celery
from htpbe_client import HtpbeError, check_pdf, decide, make_client
from storage import uploads, alerts # your code
celery = Celery("verification", broker="redis://localhost:6379/0")
celery.conf.task_acks_late = True # a crashed worker redelivers the task
# Codes that mean the integration is misconfigured or out of credits.
ALERT_CODES = {"payment_required", "invalid_api_key", "inactive_client", "test_url_required"}
@celery.task(bind=True, max_retries=8)
def verify_upload(self, upload_id: str, url: str) -> None:
try:
with make_client() as client:
result = check_pdf(client, url, idempotency_key=f"upload-{upload_id}")
except HtpbeError as exc:
if exc.is_retryable and self.request.retries < self.max_retries:
delay = exc.retry_after or min(30 * 2**self.request.retries, 900)
raise self.retry(exc=exc, countdown=delay)
uploads.save_failure(upload_id, error_code=exc.code)
if exc.code in ALERT_CODES:
alerts.page(f"HTPBE {exc.code}: verification is blocked")
return
uploads.save_verdict(
upload_id,
check_id=result.id,
state=decide(result, claims_institutional_origin=uploads.is_institutional(upload_id)),
markers=list(result.markers),
)How the two retry layers divide the work:
- Inside
check_pdf, the loop from Step 1 absorbs blips: a dropped connection, a429with a shortRetry-After. It finishes in seconds and never leaves the process. - In Celery,
self.retryhandles what outlasts that: a409that keeps coming back, a run of5xxresponses. The task goes back on the queue instead of holding a worker for minutes. A429already carries its ownRetry-After, so that value wins over the exponential countdown here too. - A worker crash is covered by
task_acks_late. The task is redelivered with the same arguments, so the same key and the same URL go out again. If the first attempt had already reached the API, the answer is the original check id and no second charge.
Permanent errors (400, 401, 402, 403, 413, 422) are written to the upload record as failed with the error_code and never retried. The ALERT_CODES set is the short list that should wake someone, because every task queued behind a 402 or 401 will fail the same way.
To test this layer without a broker, set celery.conf.task_always_eager = True and point the task at the test URLs from Step 5, such as https://api.htpbe.tech/v1/test/modified-high.pdf, with the test key in the environment. trigger-402.pdf exercises the payment_required branch end to end without touching a real balance. Keep assertions about retry timing in the httpx.MockTransport tests from Step 5, where the clock is under your control.
What this does not detect
The API reports structural evidence that a PDF was changed after it was created. It does not catch:
- A document fabricated from scratch in ordinary software and exported to PDF. There is no earlier version for it to differ from. That is what
inconclusiveis for. - Visual or content differences that leave no structural trace.
- The cryptographic validity of a signature. It detects that a signature was removed or that content follows it, not whether the certificate chain verifies.
- Password-protected files. They return
422 invalid_pdfuntil the password is removed.
Treat the verdict as one signal in front of a human reviewer, not as a ruling on the sender.
Decisions before you ship
- Sync or background. Inline in a request handler, a check costs the caller 2–5 seconds for ordinary documents. Fine for onboarding forms with a spinner; for uploads that fan out, enqueue the check and notify when it completes.
- Derive the idempotency key from your own data. A random key protects one call. If a job can be retried by a worker, a cron or a person, pass
idempotency_key=f"upload-{upload_id}"so every retry of the same upload maps to one analysis. Use a new key per document, never per run, and keep theurlidentical across those retries: the key is bound to the URL it was first sent with, which is why Step 6 mints the presigned URL once. - Which documents count as institutional.
claims_institutional_originis your policy input. Bank statements, payslips and tax certificates usually qualify; a résumé does not. - What reviewers see. Show them
status, the marker ids with their labels from the dictionary, andproducerandcreatorwhen present. Keep theresult.idin the audit log.
Same API, other stacks
The contract is identical in every language: Go, Node.js and Laravel / PHP have their own guides. For thousands of documents, queue and worker patterns cover backpressure and backfill.
The HTPBE? API is available on self-serve plans and as pay-per-check credits (see current pricing). Test keys are free, so you can build and test the whole integration before your first live check. Get your API key.
For a lighter read on the same intake idea, the April Fools’ post HTPBE? Quantum explains, after the joke, what the real check does with an invoice the day it arrives.