How Do You Write PyMongo Validation Wrapper Scripts for Safe Bulk Writes?

When you enforce a strict $jsonSchema validator across a production MongoDB cluster, a raw insert_many or bulk_write from PyMongo surfaces failures as an opaque WriteError (code 121) or a BulkWriteError that aborts the batch, leaving you guessing which document broke the contract. This page is a complete, runnable playbook for wrapping those calls: it sits inside the Python integration for schema checks workflow, which is itself the control plane of the broader Automated Schema Enforcement & Monitoring framework. By the end you will have a wrapper that runs a client-side pre-flight gate, parses the exact server-side rejection, and routes failures deterministically instead of dropping a whole batch.

A validation wrapper is the seam between application logic and the collection-level validators that MongoDB evaluates synchronously on every write. Get it right and schema drift is intercepted before it corrupts a downstream aggregation; get it wrong and one malformed document in a 10,000-document load silently discards the other 9,999.

Operational Mechanics and Write-Path Impact

A wrapper’s behavior is governed by three dials, and each interacts with the collection’s server-side enforcement differently. The client-side pre-flight (using the jsonschema library) never touches the database; the ordered flag decides whether one rejection aborts the batch; and bypass_document_validation decides whether the server validator runs at all. The table below is the decision matrix you tune against:

PyMongo setting	Value	Write-path effect	When to use
ordered	True (default)	Halts on the first rejection; remaining ops discarded	Never for bulk validation loads — one bad doc drops the rest
ordered	False	Valid docs commit; failures isolated per array index	All bulk validation pipelines
bypass_document_validation	False (default)	Server $jsonSchema validator runs; rejects with code 121	Always, so MongoDB is the final enforcement boundary
bypass_document_validation	True	Server validator skipped entirely	Never in a wrapper — it defeats the contract
pre-flight Draft7Validator	enabled	Rejects malformed payloads before the driver call	Always, to turn opaque 121s into typed local errors

Two server-side behaviors shape how the wrapper must be written. First, validationLevel: "moderate" only checks inserts and updates to documents that already satisfy the schema, so a wrapper that assumes full enforcement will silently let malformed updates through against legacy documents — the tradeoff is covered in strict vs moderate validation levels. Second, aggregation stages that write to a collection with $merge or $out bypass the $jsonSchema validator entirely, so any migration that lands documents through a pipeline needs an independent compliance count afterward. For the exact keyword semantics your pre-flight schema must mirror, see understanding MongoDB $jsonSchema syntax.

Exact Diagnostic Fingerprints and Fast Resolution

The signature failure is pymongo.errors.WriteError with code: 121 (or pymongo.errors.BulkWriteError for a batch). On MongoDB 5.0+, the driver payload carries a structured errInfo object reachable via exc.details that pinpoints the failing rule and path:

{
  "code": 121,
  "errmsg": "Document failed validation",
  "errInfo": {
    "failingDocumentId": {"$oid": "..."},
    "details": {
      "operatorName": "$jsonSchema",
      "schemaRulesNotSatisfied": [
        {
          "operatorName": "required",
          "specifiedAs": {"required": ["tenant_id", "event_ts"]},
          "missingProperties": ["event_ts"]
        }
      ]
    }
  }
}

For a BulkWriteError, the per-document failures live in exc.details["writeErrors"], each entry carrying its own index (position in the ops list), code, and errInfo. This copy-paste snippet extracts a clean, per-index failure report from either exception type:

from pymongo.errors import BulkWriteError, WriteError


def explain_validation_failure(exc):
    """Return a list of {index, missing, reason} from a Write/BulkWriteError."""
    reports = []
    write_errors = (
        exc.details.get("writeErrors", [])
        if isinstance(exc, BulkWriteError)
        else [exc.details]
    )
    for we in write_errors:
        rules = (
            we.get("errInfo", {})
            .get("details", {})
            .get("schemaRulesNotSatisfied", [])
        )
        for rule in rules:
            reports.append({
                "index": we.get("index"),
                "operator": rule.get("operatorName"),
                "missing": rule.get("missingProperties"),
                "reason": rule.get("reason"),
            })
    return reports

Three fingerprints account for the majority of production incidents:

• code: 121 on a document you believe is valid is almost always a bsonType mismatch — a JSON integer is stored as int/long, not double, so a schema demanding "double" rejects it. Widen to bsonType: ["double", "int", "long", "decimal"].
• A whole batch reported as failed usually means ordered=True aborted after the first rejection; only the first writeErrors entry is real, the rest never executed.
• Documents that bypassed the validator entirely come from a $merge/$out migration. Confirm with a post-hoc count: db.dest.countDocuments({"$nor": [{"$jsonSchema": schema}]}).

Step-by-Step Playbook

The wrapper decouples schema evaluation from write execution: a synchronous pre-flight gate rejects malformed payloads locally, then the valid subset is written with ordered=False so failures isolate per index. Follow these four steps.

1. Build the wrapper with a client-side pre-flight gate. The Draft7Validator mirrors the server schema so bad documents never reach the driver:

import jsonschema
from jsonschema import Draft7Validator
from pymongo import MongoClient, InsertOne, errors
from typing import Dict, List, Any


class ValidationWrapper:
    def __init__(self, client: MongoClient, db_name: str,
                 collection_name: str, schema: Dict[str, Any]):
        self.collection = client[db_name][collection_name]
        self.schema = schema
        self.validator = Draft7Validator(schema)

    def is_valid(self, doc: Dict[str, Any]) -> bool:
        """True if the document satisfies the schema; False otherwise."""
        try:
            self.validator.validate(doc)
            return True
        except jsonschema.ValidationError:
            return False

2. Partition the batch and write the valid subset unordered. Keep the invalid documents so they can be routed, not lost:

    def safe_bulk_insert(self, documents: List[Dict[str, Any]]) -> Dict[str, Any]:
        valid = [d for d in documents if self.is_valid(d)]
        invalid = [d for d in documents if d not in valid]

        if not valid:
            return {"inserted": 0, "skipped": len(invalid), "invalid": invalid}

        try:
            result = self.collection.bulk_write(
                [InsertOne(d) for d in valid],
                ordered=False,
                bypass_document_validation=False,
            )
            return {"inserted": result.inserted_count,
                    "skipped": len(invalid), "invalid": invalid}
        except errors.BulkWriteError as bwe:
            failed = explain_validation_failure(bwe)
            return {"inserted": bwe.details.get("nInserted", 0),
                    "skipped": len(invalid), "server_rejected": failed}

3. Route rejected documents to a dead-letter queue. Preserve the original payload, a timestamp, and the parsed error signature so remediation is asynchronous and non-blocking. This is the entry point into your fallback validation chains, and tagging each failure by rule feeds directly into categorizing schema validation errors:

    def to_dead_letter(self, dlq, invalid_docs, signatures):
        import datetime
        dlq.insert_many([
            {"payload": doc, "ts": datetime.datetime.utcnow(),
             "error": sig}
            for doc, sig in zip(invalid_docs, signatures)
        ])

4. Verify no documents slipped past the collection validator. Because $jsonSchema is a valid query operator, $nor counts non-compliant documents directly — run this after every load and especially after any $merge/$out migration:

    def compliance_gap(self) -> int:
        """Documents in the collection that violate the schema (should be 0)."""
        return self.collection.count_documents(
            {"$nor": [{"$jsonSchema": self.schema}]}
        )

Expected output for a healthy load is {"inserted": 9997, "skipped": 3, ...} with compliance_gap() == 0. A non-zero gap means documents entered through a validator-bypassing path and need reconciliation. Anchoring the wrapper’s schema to a version-controlled registry — aligned with your schema versioning strategies — keeps the client pre-flight and server validator from drifting apart.

Failure Modes & Rollback

Each step has a distinct failure mode and a bounded recovery path:

• Pre-flight schema drifts from the server validator. The wrapper passes documents the server then rejects, producing surprise BulkWriteErrors. Recover: re-pull the schema from the registry into the Draft7Validator; time-to-recover is a redeploy of the wrapper, seconds to minutes.
• ordered=True left in place. A single bad document aborts the batch and the run looks like a total failure. Recover: set ordered=False and re-run; only the previously-skipped documents remain, and inserts are idempotent if you supply stable _ids.
• Enforcement flip storms. Promoting the collection from validationAction: "warn" to "error" on top of legacy data turns a routine load into a write outage. Rollback: db.runCommand({ collMod: "events", validationAction: "warn" }) restores writes instantly while keeping the schema attached; time-to-recover is the replication lag of your slowest secondary (typically sub-second). Promote back to error only once the wrapper’s compliance_gap() and your async validation monitoring dashboards both read zero.
• DLQ backlog grows unbounded. A schema-versioning bug can flood the dead-letter collection. Recover: the DLQ is append-only and non-destructive, so replay is a batch re-validate against the corrected schema; nothing is lost.

Frequently Asked Questions

Related

• Python integration for schema checks — the parent workflow this wrapper plugs into, covering the full read-diff-apply control loop for validator changes.
• Implementing collection-level validators — how the server-side $jsonSchema gate the wrapper writes against is deployed and rolled back.
• Building fallback validation chains — where dead-lettered documents go so a strict validator never means data loss.
• Categorizing schema validation errors — turning each parsed code-121 signature into a bug, migration gap, or schema-change signal.
• Async validation monitoring dashboards — the telemetry that tells you when the compliance gap is zero and enforcement is safe to promote.