DEV Community

Dave Sng
Dave Sng

Posted on

Stop Bad Data Before It Reaches Your Database: API Input Validation Patterns in Python

#python #webdev #api #tutorial

Garbage in, garbage out. It's an old saying, but it still describes one of the most common causes of production bugs in modern APIs.

Whether you're building a fintech app, an e-commerce platform, or a SaaS product, bad input data will find its way into your system. The question is whether you catch it at the API boundary — or after it's corrupted your database.

In this guide, we'll walk through practical Python patterns for API input validation, from the basics to more advanced techniques used in production systems.

Why Input Validation Matters More Than Ever

In 2026, APIs are no longer just integration glue — they're regulated infrastructure. Financial APIs, healthcare APIs (FHIR), and open banking platforms are increasingly subject to compliance audits that require demonstrating where and how data is validated.

Bad data causes:

  • Silent data corruption — nulls, wrong types, or out-of-range values that slip into the database
  • Security vulnerabilities — injection attacks, prototype pollution, and schema-breaking payloads
  • Compliance failures — regulators increasingly expect APIs to validate and log what they accept

The cost of fixing bad data downstream is orders of magnitude higher than catching it at the API boundary.

Layer 1: Schema Validation with Pydantic

The first line of defense in any Python API is schema validation. Pydantic is the gold standard — it's built into FastAPI and validates Python type hints at runtime.

from pydantic import BaseModel, EmailStr, field_validator
from typing import Optional
from datetime import date

class TransactionRequest(BaseModel):
    amount: float
    currency: str
    payer_email: EmailStr
    transaction_date: date
    reference_id: Optional[str] = None

    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v):
        allowed = {"USD", "EUR", "GBP", "MYR", "SGD"}
        if v.upper() not in allowed:
            raise ValueError(f"Currency must be one of: {allowed}")
        return v.upper()

    @field_validator("amount")
    @classmethod
    def validate_amount(cls, v):
        if v <= 0:
            raise ValueError("Amount must be positive")
        if v > 1_000_000:
            raise ValueError("Amount exceeds single-transaction limit")
        return round(v, 2)
Enter fullscreen mode Exit fullscreen mode

With FastAPI, this model auto-generates an OpenAPI schema and returns structured 422 errors if validation fails — no extra code needed:

from fastapi import FastAPI

app = FastAPI()

@app.post("/transactions")
async def create_transaction(payload: TransactionRequest):
    # At this point, payload is guaranteed to be valid
    return {"status": "accepted", "reference": payload.reference_id}
Enter fullscreen mode Exit fullscreen mode

Layer 2: Business Rule Validation

Schema validation ensures types are correct. Business rule validation ensures the logic is sound.

from fastapi import HTTPException

async def validate_business_rules(payload: TransactionRequest):
    # Rule 1: Reject transactions on weekends for regulated accounts
    if payload.transaction_date.weekday() >= 5:
        raise HTTPException(
            status_code=422,
            detail="Weekend transactions not permitted for regulated accounts"
        )

    # Rule 2: Enforce additional checks for large non-USD amounts
    if payload.amount > 50_000 and payload.currency != "USD":
        raise HTTPException(
            status_code=422,
            detail="Large non-USD transactions require additional verification"
        )
Enter fullscreen mode Exit fullscreen mode

Business rules are where most teams fall short. They handle schema validation but skip the domain-specific checks that prevent real-world fraud and compliance failures.

Layer 3: External Data Validation APIs

Some validation is too complex or too data-intensive to build in-house. This is where specialized validation APIs shine.

For example, validating whether a dataset conforms to expected formats, checking for structural anomalies, or running multi-field cross-validation rules across large payloads — these are tasks that benefit from purpose-built tooling.

DataForge is a data validation API on RapidAPI that handles exactly this: it takes raw data payloads and runs configurable validation pipelines, returning detailed field-level error reports. This is particularly useful when you're ingesting data from external partners or user uploads where the structure is unpredictable.

import httpx

DATAFORGE_KEY = "your-rapidapi-key"
DATAFORGE_HOST = "dataforge2.p.rapidapi.com"

async def validate_with_dataforge(data: dict) -> dict:
    """Run external validation pipeline on incoming data."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://{DATAFORGE_HOST}/validate",
            headers={
                "X-RapidAPI-Key": DATAFORGE_KEY,
                "X-RapidAPI-Host": DATAFORGE_HOST,
                "Content-Type": "application/json",
            },
            json={
                "payload": data,
                "rules": ["required_fields", "type_check", "range_check"],
            },
            timeout=10.0,
        )
        return response.json()

# Usage in your endpoint
@app.post("/ingest")
async def ingest_data(payload: dict):
    result = await validate_with_dataforge(payload)

    if not result.get("valid"):
        raise HTTPException(
            status_code=422,
            detail={
                "message": "Data validation failed",
                "errors": result.get("errors", []),
            }
        )

    return {"status": "ingested", "record_count": result.get("record_count")}
Enter fullscreen mode Exit fullscreen mode

The key advantage: your application code stays clean, and you can update validation rules through the API without redeploying.

Layer 4: Returning Structured Error Responses

Good validation is useless if your error responses are cryptic. Clients need actionable feedback.

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " -> ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
        })

    return JSONResponse(
        status_code=422,
        content={
            "status": "validation_error",
            "errors": errors,
            "hint": "Check field types and required values against the API schema.",
        }
    )
Enter fullscreen mode Exit fullscreen mode

Sample response a client would receive:

{
  "status": "validation_error",
  "errors": [
    {
      "field": "body -> currency",
      "message": "Currency must be one of: {'USD', 'EUR', 'GBP'}",
      "type": "value_error"
    },
    {
      "field": "body -> amount",
      "message": "Amount must be positive",
      "type": "value_error"
    }
  ],
  "hint": "Check field types and required values against the API schema."
}
Enter fullscreen mode Exit fullscreen mode

Layer 5: Logging Validation Failures for Compliance

In regulated industries, it's not enough to reject bad data — you need to record what was rejected and why.

import logging
import json
from datetime import datetime, timezone

logger = logging.getLogger("validation_audit")

async def log_validation_failure(
    endpoint: str,
    errors: list,
    client_id: str | None = None,
):
    logger.warning(json.dumps({
        "event": "validation_failure",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "endpoint": endpoint,
        "client_id": client_id or "anonymous",
        "error_count": len(errors),
        "errors": errors,
    }))
Enter fullscreen mode Exit fullscreen mode

Ship these logs to your SIEM or observability platform. During a compliance audit, this trail demonstrates that your API actively enforces data quality — not just hopeful trust.

Putting It All Together

Here's the validation stack in a single FastAPI endpoint:

@app.post("/transactions")
async def create_transaction(
    request: Request,
    payload: TransactionRequest,  # Layer 1: schema
):
    # Layer 2: business rules
    await validate_business_rules(payload)

    # Layer 3: external validation
    ext_result = await validate_with_dataforge(payload.model_dump())
    if not ext_result.get("valid"):
        await log_validation_failure(
            endpoint="/transactions",
            errors=ext_result.get("errors", []),
            client_id=request.headers.get("X-Client-ID"),
        )
        raise HTTPException(status_code=422, detail=ext_result["errors"])

    # All layers passed — safe to persist
    return {"status": "accepted"}
Enter fullscreen mode Exit fullscreen mode

Key Takeaways

  1. Validate at the boundary — never trust data that hasn't been checked
  2. Use schema validation as the first gate — Pydantic + FastAPI makes this nearly free
  3. Layer business rules on top — domain logic catches what schema validation misses
  4. Offload complex validation to specialized APIs when rules are complex or data-intensive
  5. Log rejections — validation audit trails are increasingly a compliance requirement

Good validation isn't just defensive coding — it's the foundation of trustworthy APIs.

Dave Sng is an API builder based in Malaysia, specializing in data validation, compliance tooling, and developer-focused APIs. He builds and publishes APIs on RapidAPI.

Top comments (0)