Integrity Failures in Django with Dynamodb

Integrity failures occur when an application fails to enforce data correctness, consistency, or trust boundaries between operations, components, or external systems. In a Django application that uses Amazon DynamoDB as its primary data store, integrity risks arise from the mismatch between Django’s relational-oriented patterns and DynamoDB’s key-value/document model. Because DynamoDB does not enforce relational constraints such as foreign keys, unique indexes across attributes, or cascading updates, responsibilities that would typically be handled by the database are shifted to the application layer.

When Django code assumes ACID-style guarantees that DynamoDB does not provide, integrity issues can emerge. For example, without explicit conditional writes or transactional checks in DynamoDB, two concurrent requests might read the same balance, compute updates independently, and write back conflicting values, resulting in a lost update. Additionally, incomplete or inconsistent multi-step workflows—such as creating a financial record and then updating a related ledger—can leave the dataset in an invalid state if error handling or rollback logic is absent.

Django’s ORM abstractions can obscure the fact that DynamoDB operates without schema-level integrity constraints. Developers might rely on model-level unique_together constraints or clean() validations, which are enforced at the Django layer and not propagated to DynamoDB. If requests bypass Django (e.g., direct SDK usage, batch jobs, or third-party integrations), those safeguards are absent, increasing the risk of invalid data. Similarly, missing implementation of conditional expressions like attribute_not_exists or version checks can allow overwrites of existing records, violating invariants such as “one primary contact per customer.”

The combination of Django’s high-level abstractions and DynamoDB’s schema flexibility also complicates auditability. Without consistent attribute typing, a numeric field intended to represent a quantity might inadvertently store a string, leading to comparison errors during authorization or pricing calculations. Inadequate handling of attribute resolution in nested structures can further introduce logic flaws where updates apply only to partial data, creating fragmented integrity violations that are difficult to trace.

Integrity failures in this context often map to the OWASP API Top 10 category BOLA/IDOR and Property Authorization, where authorization checks are incomplete or data boundaries are not enforced. They can also relate to Input Validation and Data Exposure if malformed payloads produce unexpected states or sensitive data is derived from inconsistent sources. Proactively validating state transitions, using conditional writes, and modeling domain rules explicitly in DynamoDB are essential to maintaining integrity in Django-Dynamodb integrations.

Remediation focuses on explicitly enforcing integrity at the DynamoDB interaction layer, since Django cannot provide it automatically. Use conditional writes, version attributes, and transactional patterns to ensure operations adhere to business rules. Below are concrete, working examples using the AWS SDK for Python (Boto3) within a Django project.

1. Prevent lost updates with conditional writes

Use a version attribute and ConditionExpression to ensure updates only succeed if the record has not changed since it was read.

import boto3
from django.core.exceptions import ValidationError

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('accounts')

def safe_update_balance(account_id, expected_version, delta):
    response = table.update_item(
        Key={'account_id': account_id},
        UpdateExpression='SET balance = balance + :delta, version = version + :inc',
        ConditionExpression='version = :expected_version',
        ExpressionAttributeValues={
            ':delta': delta,
            ':inc': 1,
            ':expected_version': expected_version
        },
        ReturnValues='UPDATED_NEW'
    )
    return response['Attributes']

# Usage in a Django view or service:
# current = table.get_item(Key={'account_id': '123'})
try:
    updated = safe_update_balance('123', expected_version=current['Item']['version'], delta=10)
except Exception as e:
    if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
        raise ValidationError('Concurrent update detected. Please retry.')
    raise

2. Enforce one-primary-contact invariant with attribute_not_exists

Ensure a uniqueness constraint that DynamoDB does not enforce natively by checking for existence before creation.

def create_primary_contact(customer_id, contact_data):
    # Check if a primary contact already exists for this customer
    existing = table.query_index(
        IndexName='gsi-primary-contact',
        KeyConditionExpression='customer_id = :cid AND contact_type = :pt',
        ExpressionAttributeValues={':cid': customer_id, ':pt': 'primary'}
    )
    if existing.get('Items'):
        raise ValidationError('A primary contact already exists for this customer.')

    # Create the item with a uniqueness guard
    table.put_item(
        Item={
            'contact_id': str(uuid4()),
            'customer_id': customer_id,
            'contact_type': 'primary',
            'email': contact_data['email'],
            'created_at': datetime.utcnow().isoformat()
        }
    )

3. Use transactions for multi-step integrity

Group related writes into a transaction to maintain consistency across multiple items.

def record_transaction_and_update_ledger(transaction_id, account_id, amount):
    table.transact_write_items(
        TransactItems=[
            {
                'Update': {
                    'TableName': 'accounts',
                    'Key': {'account_id': {'S': account_id}},
                    'UpdateExpression': 'SET balance = balance + :amt',
                    'ConditionExpression': 'balance >= :min_balance',
                    'ExpressionAttributeValues': {
                        ':amt': {'N': str(amount)},
                        ':min_balance': {'N': '0'}
                    }
                }
            },
            {
                'Put': {
                    'TableName': 'transaction_log',
                    'Item': {
                        'transaction_id': {'S': transaction_id},
                        'account_id': {'S': account_id},
                        'amount': {'N': str(amount)},
                        'timestamp': {'S': datetime.utcnow().isoformat()}
                    }
                }
            }
        ]
    )

4. Validate and coerce attribute types

Ensure numeric fields remain numeric and enforce domain rules before writing.

def prepare_order_item(item):
    if not isinstance(item.get('quantity'), int) or item['quantity'] <= 0:
        raise ValidationError('Quantity must be a positive integer.')
    if not isinstance(item.get('unit_price'), (int, float)) or item['unit_price'] < 0:
        raise ValidationError('Unit price must be a non-negative number.')
    return {
        'order_id': {'S': item['order_id']},
        'product_id': {'S': item['product_id']},
        'quantity': {'N': str(item['quantity'])},
        'unit_price': {'N': str(item['unit_price'])},
        'total': {'N': str(item['quantity'] * item['unit_price'])}
    }

By combining conditional expressions, transactions, explicit uniqueness checks, and strict input validation, Django applications can maintain data integrity when using DynamoDB. These patterns address the lack of native relational guarantees and ensure that business rules are enforced consistently at the point of data access.