← Back to DevBytes

Fix 'AssertionError' in Python in Production: Root Cause Analysis

Understanding AssertionError in Python

An AssertionError is raised when an assert statement fails. The assert keyword is a debugging aid that tests a condition as a sanity check. If the condition evaluates to True, execution continues normally. If it evaluates to False, Python raises an AssertionError — potentially with an optional message you provide.

Here is the basic syntax:

assert condition, "Optional descriptive message"

And a simple example that triggers the error:

def calculate_discount(price, percentage):
    assert percentage >= 0, "Discount percentage cannot be negative"
    assert percentage <= 100, "Discount percentage cannot exceed 100"
    return price * (1 - percentage / 100)

# This works fine
print(calculate_discount(200, 20))  # 160.0

# This blows up
print(calculate_discount(200, -5))  # AssertionError: Discount percentage cannot be negative

On the surface, AssertionError looks like a simple runtime exception. In production, however, it signals something far more dangerous: an invariant violation — a condition the developer believed would always hold true, but which reality has disproved.

Why AssertionError Matters in Production

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

When an assert fails in a live system, it is not just a bug — it is a broken assumption. The developer wrote that assertion because they were certain something could never happen. Production just proved them wrong. The consequences range from crashed services to silently corrupted data, depending on how the error is handled (or not handled).

Consider these real-world scenarios where production AssertionErrors can wreak havoc:

The key insight is this: an AssertionError in production is never about the assertion itself — it is about the flawed assumption that created it. Root cause analysis must target that assumption, not just silence the error.

Root Cause Analysis: A Systematic Approach

Fixing an AssertionError in production requires moving beyond the stack trace. Here is a step-by-step methodology you can apply immediately when an assertion fails in a live environment.

Step 1: Capture the Full Context

Before you touch a single line of code, gather forensic data. The assertion message (if present) and the stack trace are your starting points, but they are rarely enough. You need:

Here is a practical decorator that captures this context automatically when an assertion fails:

import functools
import logging
import traceback
import json
from datetime import datetime

logger = logging.getLogger("assertion_forensics")

def capture_assertion_context(func):
    """
    Decorator that captures local variables and inputs when an
    AssertionError occurs, then re-raises with enriched information.
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except AssertionError as original_error:
            # Capture local variables from the frame where the assert failed
            tb = traceback.extract_tb(original_error.__traceback__)
            # The last frame is where the assertion actually failed
            failing_frame = tb[-1]
            
            context = {
                "timestamp": datetime.utcnow().isoformat(),
                "function": func.__name__,
                "failing_file": failing_frame.filename,
                "failing_line": failing_frame.lineno,
                "failing_code": failing_frame.line,
                "assertion_message": str(original_error),
                "positional_args_preview": repr(args[:5]),  # first 5 for sanity
                "keyword_args_keys": list(kwargs.keys()),
            }
            
            logger.error(
                f"AssertionError captured | Context: {json.dumps(context, indent=2)}"
            )
            # Re-raise so normal error handling still fires
            raise
    return wrapper

# Usage example
@capture_assertion_context
def transfer_funds(sender_balance, amount):
    assert sender_balance >= amount, \
        f"Insufficient funds: balance={sender_balance}, amount={amount}"
    new_balance = sender_balance - amount
    assert new_balance >= 0, \
        f"Post-transfer balance went negative: {new_balance}"
    return new_balance

With this decorator in place, every AssertionError leaves a detailed forensic footprint in your logs — giving you far more to work with than a naked stack trace.

Step 2: Reproduce the Failure in Isolation

Once you have the inputs that triggered the failure, reproduce it in a controlled environment. Do not skip this step. The goal is to confirm that the assertion fails deterministically with those inputs and to understand exactly which condition broke.

# Reproduction script — isolate the failing logic
def reproduce_assertion_failure():
    # These values came from production logs
    sender_balance = 100.00
    amount = 100.01  # The 0.01 discrepancy that broke the assertion
    
    try:
        result = transfer_funds(sender_balance, amount)
        print(f"Result: {result}")
    except AssertionError as e:
        print(f"Reproduced: {e}")
        # Now you can inspect, add print statements, attach a debugger
        import pdb; pdb.set_trace()

reproduce_assertion_failure()

During reproduction, ask these questions:

Step 3: Classify the Root Cause

Every production AssertionError falls into one of these categories. Identifying the correct category dictates the fix.

Category A — Invalid Assumption About Inputs: The developer assumed inputs would always satisfy certain constraints, but external data sources, user input, or upstream services violated those constraints.

Category B — Logic Error in the Code Itself: The assertion guards against a condition that the surrounding code should prevent, but a bug in that surrounding code allows the condition to occur anyway.

Category C — Environmental or State Corruption: The assertion depends on external state (file system, database, in-memory cache, environment variables) that changed unexpectedly between when the assumption was made and when the assertion ran.

Category D — Concurrency or Timing Issue: The assertion assumes atomicity or ordering that does not hold under concurrent access patterns, race conditions, or eventual consistency models.

Here is how you might instrument code to distinguish these categories:

def process_order(order, inventory, user_account):
    # Category A example: invalid assumption about input structure
    assert "items" in order, \
        f"Order missing 'items' key: order_id={order.get('id')}"
    
    # Category B example: logic should guarantee this, but a bug might let it slip
    total_amount = sum(item["price"] * item["quantity"] for item in order["items"])
    assert total_amount > 0, \
        f"Total amount must be positive, got {total_amount} for order {order['id']}"
    
    # Category C example: environment-dependent invariant
    assert inventory.get_available_stock(order["items"]) is not None, \
        f"Inventory service returned None — possible outage or config change"
    
    # Category D example: concurrency assumption
    # (Assume this is inside a lock, but the lock might not cover this section)
    assert user_account.balance >= total_amount, \
        f"Balance check failed: balance={user_account.balance}, total={total_amount}"
    
    # ... rest of processing

Step 4: Choose the Right Fix

Once you have classified the root cause, the fix becomes clear. Do not simply remove the assertion. Here is what to do for each category:

For Category A (Invalid Input Assumptions):

# BEFORE — fragile assertion on input
def handle_webhook(payload):
    assert payload["event_type"] in ("payment.success", "payment.failed")
    # process...

# AFTER — explicit validation with graceful degradation
from enum import Enum

class WebhookEventType(Enum):
    PAYMENT_SUCCESS = "payment.success"
    PAYMENT_FAILED = "payment.failed"

class InvalidWebhookError(ValueError):
    """Raised when webhook payload does not match expected schema."""
    pass

def handle_webhook(payload):
    try:
        event_type = WebhookEventType(payload.get("event_type"))
    except (ValueError, KeyError) as e:
        raise InvalidWebhookError(
            f"Unrecognized webhook event: {payload.get('event_type')}"
        ) from e
    # Now event_type is guaranteed to be valid
    process_event(event_type, payload)

For Category B (Logic Errors):

# The assertion revealed a logic bug — fix the root cause, keep the assertion
def apply_discounts(cart_items, promotions):
    # BUG: The original code applied promotions in the wrong order,
    # allowing a negative subtotal to sneak through.
    # FIX: Apply percentage discounts first, then flat discounts,
    # with a floor of zero at each step.
    subtotal = sum(item.price for item in cart_items)
    
    # Apply percentage discounts
    for promo in promotions:
        if promo.type == "percentage":
            discount = subtotal * (promo.value / 100)
            subtotal = max(subtotal - discount, 0)  # Floor at zero
    
    # Apply flat discounts
    for promo in promotions:
        if promo.type == "flat":
            subtotal = max(subtotal - promo.value, 0)
    
    # This assertion now guards against future regressions
    assert subtotal >= 0, f"Subtotal became negative: {subtotal}"
    return subtotal

For Category C (Environmental / State Corruption):

# BEFORE — assertion that fails when Redis is temporarily unreachable
def get_user_session(session_id, redis_client):
    session_data = redis_client.get(session_id)
    assert session_data is not None, f"Session {session_id} not found in Redis"
    return deserialize(session_data)

# AFTER — graceful handling with fallback and monitoring
class SessionNotFoundError(Exception):
    pass

class CacheUnavailableError(Exception):
    pass

def get_user_session(session_id, redis_client, fallback_store=None):
    try:
        session_data = redis_client.get(session_id)
    except ConnectionError as e:
        raise CacheUnavailableError(
            f"Redis unreachable for session {session_id}"
        ) from e
    
    if session_data is None:
        if fallback_store:
            session_data = fallback_store.get(session_id)
        if session_data is None:
            raise SessionNotFoundError(
                f"Session {session_id} not found in primary or fallback storage"
            )
    
    return deserialize(session_data)

For Category D (Concurrency / Timing Issues):

# BEFORE — assertion that fails under race conditions
counter = 0

def increment():
    global counter
    current = counter
    # Gap here — another thread can modify counter
    counter = current + 1
    assert counter == current + 1, "Counter invariant broken"

# AFTER — proper locking or atomic operations
import threading

counter = 0
counter_lock = threading.Lock()

def increment():
    global counter
    with counter_lock:
        current = counter
        counter = current + 1
        # Now this assertion is genuinely safe
        assert counter == current + 1, "Counter invariant broken"

# EVEN BETTER — use atomic types from the standard library
from threading import Lock

class AtomicCounter:
    def __init__(self):
        self._value = 0
        self._lock = Lock()
    
    def increment(self):
        with self._lock:
            self._value += 1
            return self._value
    
    @property
    def value(self):
        with self._lock:
            return self._value

Production-Safe Assertion Patterns

Now that you understand root cause analysis, here are practical patterns to prevent AssertionError surprises in production altogether.

Pattern 1: Conditional Assertions (Dev vs. Production)

Some assertions are genuinely only useful during development. Use an environment-aware wrapper that strips them in production or converts them to warnings.

import os
import warnings

ENVIRONMENT = os.environ.get("APP_ENV", "development")

def safe_assert(condition, message=""):
    """
    In development: behaves like a normal assert (raises AssertionError).
    In production: emits a warning instead of crashing.
    """
    if not condition:
        if ENVIRONMENT == "development":
            raise AssertionError(message)
        else:
            warnings.warn(
                f"Assertion would have failed: {message}",
                RuntimeWarning
            )

# Usage
def calculate_shipping(weight_kg, destination):
    safe_assert(weight_kg > 0, f"Weight must be positive, got {weight_kg}")
    safe_assert(destination in SUPPORTED_COUNTRIES,
                f"Unsupported destination: {destination}")
    # ... calculation logic

Pattern 2: Assertions as Documentation with Graceful Fallback

Use assertions to document invariants in your codebase, but pair every assertion with a corresponding runtime check that handles the failure case gracefully. This is the "belt and suspenders" approach.

class TemperatureSensor:
    """
    Reads temperature from a hardware sensor.
    
    Invariant: readings are always between -50°C and 150°C
    (the physical operating range of the sensor hardware).
    """
    
    MIN_VALID_TEMP = -50.0
    MAX_VALID_TEMP = 150.0
    
    def read_temperature(self):
        raw_value = self._read_raw_sensor()
        
        # Documentation assertion — this should never fail
        assert self.MIN_VALID_TEMP <= raw_value <= self.MAX_VALID_TEMP, \
            f"Raw sensor reading {raw_value} out of physical range"
        
        # Graceful fallback — handles the "impossible" case anyway
        if raw_value < self.MIN_VALID_TEMP or raw_value > self.MAX_VALID_TEMP:
            # Log as a critical anomaly, then clamp to a safe default
            logger.critical(
                f"Sensor anomaly detected: {raw_value}°C — clamping to safe range"
            )
            return self._estimate_temperature_from_neighbors()
        
        return raw_value

Pattern 3: Structured Assertion Helpers for Complex Invariants

For data structures with multiple invariants, build dedicated validation functions that provide detailed diagnostics when an invariant breaks. This makes root cause analysis dramatically faster.

class OrderValidator:
    """Validates all invariants for an Order object and provides detailed diagnostics."""
    
    @staticmethod
    def validate(order):
        violations = []
        
        # Invariant 1: Order must have at least one line item
        if not order.items or len(order.items) == 0:
            violations.append({
                "invariant": "non_empty_items",
                "expected": "at least 1 item",
                "actual": len(order.items) if order.items else 0,
                "order_id": order.id
            })
        
        # Invariant 2: Total amount must match sum of line items (within rounding tolerance)
        computed_total = sum(
            item.unit_price * item.quantity for item in order.items
        )
        tolerance = 0.01  # 1 cent tolerance for rounding
        if abs(order.total_amount - computed_total) > tolerance:
            violations.append({
                "invariant": "total_amount_consistency",
                "expected_total": computed_total,
                "actual_total": order.total_amount,
                "difference": order.total_amount - computed_total,
                "order_id": order.id
            })
        
        # Invariant 3: Status must be a valid transition from the previous status
        valid_transitions = {
            "pending": ["confirmed", "cancelled"],
            "confirmed": ["shipped", "cancelled"],
            "shipped": ["delivered"],
            "delivered": [],
            "cancelled": []
        }
        if order.previous_status:
            allowed = valid_transitions.get(order.previous_status, [])
            if order.status not in allowed:
                violations.append({
                    "invariant": "valid_status_transition",
                    "from_status": order.previous_status,
                    "to_status": order.status,
                    "allowed_transitions": allowed,
                    "order_id": order.id
                })
        
        if violations:
            raise OrderInvariantViolation(
                f"Order {order.id} violates {len(violations)} invariants",
                violations=violations
            )
        return True

class OrderInvariantViolation(Exception):
    def __init__(self, message, violations):
        super().__init__(message)
        self.violations = violations

# Usage in production code
def process_order(order):
    try:
        OrderValidator.validate(order)
    except OrderInvariantViolation as e:
        logger.error(
            f"Order invariant violation: {e.violations}",
            extra={"order_id": order.id}
        )
        # Route to manual review queue instead of crashing
        send_to_review_queue(order, reason=str(e.violations))
        return
    # Normal processing continues...

Best Practices for Assertions in Production Code

Real-World Example: Debugging a Production AssertionError End-to-End

Let's walk through a complete, realistic scenario that ties together everything covered so far.

The incident: At 3:14 AM, your payment processing service starts crashing with AssertionError: Refund amount must not exceed original payment amount. The on-call engineer sees a cascade of failed health checks and rolls back to the previous deployment, which stops the bleeding. Now it is your job to perform root cause analysis.

The offending code:

def process_refund(payment_record, refund_amount, reason):
    # Invariant: you cannot refund more than the original payment
    assert refund_amount <= payment_record.amount, \
        f"Refund amount {refund_amount} exceeds original payment {payment_record.amount}"
    
    # Invariant: refund reason must be non-empty
    assert reason and len(reason.strip()) > 0, \
        "Refund reason cannot be empty"
    
    # Proceed with refund via payment gateway
    gateway_response = payment_gateway.refund(
        transaction_id=payment_record.gateway_id,
        amount=refund_amount,
        reason=reason
    )
    
    assert gateway_response.status == "success", \
        f"Gateway refund failed: {gateway_response.error}"
    
    return gateway_response

Step 1 — Forensic data: The logs show the failing call had payment_record.amount = 99.99 and refund_amount = 99.99. Wait — they are equal, so the assertion refund_amount <= payment_record.amount should have passed. Something is wrong with our understanding.

Step 2 — Reproduction:

# Attempting to reproduce
payment_record_amount = 99.99
refund_amount = 99.99
print(refund_amount <= payment_record_amount)  # Expected: True, but...

# Let's check the actual types
print(type(payment_record_amount))  # 
print(type(refund_amount))          # 

# Floating point precision check
print(f"{payment_record_amount:.20f}")  # 99.98999999999999488409...
print(f"{refund_amount:.20f}")          # 99.99000000000000909494...
print(refund_amount <= payment_record_amount)  # False! Due to IEEE 754 drift

There it is. The payment_record.amount came from a database DECIMAL(10,2) column that was cast to a Python float during ORM mapping, introducing a tiny precision loss. The refund_amount came from a JSON API request parsed as a float with a slightly different binary representation. The assertion that compared them with <= failed due to floating-point arithmetic, not due to an actual business logic violation.

Step 3 — Root cause classification: This is Category C (Environmental / State Corruption) combined with a subtle data representation issue. The assumption that two floating-point numbers representing the same decimal value will compare equal or in-order is fundamentally invalid.

Step 4 — The fix:

from decimal import Decimal, ROUND_HALF_UP

def process_refund(payment_record, refund_amount, reason):
    # Convert both amounts to Decimal for exact decimal comparison
    original_amount = Decimal(str(payment_record.amount)).quantize(
        Decimal("0.01"), rounding=ROUND_HALF_UP
    )
    requested_refund = Decimal(str(refund_amount)).quantize(
        Decimal("0.01"), rounding=ROUND_HALF_UP
    )
    
    # Now the comparison is exact and predictable
    if requested_refund > original_amount:
        raise RefundValidationError(
            f"Refund amount {requested_refund} exceeds original payment "
            f"{original_amount}",
            original_amount=original_amount,
            refund_amount=requested_refund
        )
    
    # Validate reason
    if not reason or len(reason.strip()) == 0:
        raise RefundValidationError(
            "Refund reason cannot be empty",
            original_amount=original_amount,
            refund_amount=requested_refund
        )
    
    # Proceed with refund
    gateway_response = payment_gateway.refund(
        transaction_id=payment_record.gateway_id,
        amount=requested_refund,
        reason=reason
    )
    
    if gateway_response.status != "success":
        raise RefundGatewayError(
            f"Gateway refund failed: {gateway_response.error}",
            gateway_response=gateway_response
        )
    
    return gateway_response

class RefundValidationError(ValueError):
    """Raised when refund request violates business rules."""
    def __init__(self, message, original_amount=None, refund_amount=None):
        super().__init__(message)
        self.original_amount = original_amount
        self.refund_amount = refund_amount

class RefundGatewayError(RuntimeError):
    """Raised when the payment gateway rejects a refund."""
    def __init__(self, message, gateway_response=None):
        super().__init__(message)
        self.gateway_response = gateway_response

This fix accomplishes several goals: it eliminates the floating-point comparison by using Decimal, it replaces bare assertions with domain-specific exceptions that carry structured context, and it preserves the invariant checks as explicit business logic that is easy to test and monitor.

Conclusion

An AssertionError in production is a gift — it exposes a gap between what you believe about your system and how it actually behaves. The root cause is never the assertion itself; it is the flawed assumption that led you to write it. By systematically capturing forensic context, reproducing the failure, classifying the root cause into one of four categories, and applying the appropriate fix — whether that means better input validation, correcting a logic error, hardening environmental dependencies, or adding proper synchronization — you transform a disruptive production crash into a permanent improvement in your system's resilience. Treat assertions as executable documentation of your invariants, but always pair them with graceful degradation paths. The goal is not to eliminate assertions from your codebase; it is to make every assertion that remains so well-understood and well-guarded that it never fires in production again.

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles