Pydantic Financial Transaction Example Explanation

This document explains a Python code example that uses Pydantic to model and validate financial transaction data, provided in pydantic_transaction_example.py.

Overview

We'll break down each part of the code and explain its purpose.

Code Explanation

Imports and Base Setup

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import List, Optional
from datetime import datetime
from decimal import Decimal
import re

This section imports necessary modules:

• Pydantic classes for data modeling and validation
• Python's typing module for type hints
• datetime for handling timestamps
• Decimal for precise financial calculations
• re for regular expression matching

Base Transaction Model

class TransactionBase(BaseModel):
    transaction_id: str = Field(..., min_length=10, max_length=50)
    amount: Decimal = Field(..., ge=Decimal('0.01'))
    timestamp: datetime

TransactionBase is the foundation for all transactions:

• transaction_id: A string between 10 and 50 characters
• amount: A decimal number greater than or equal to 0.01
• timestamp: The date and time of the transaction

Custom Validators

    @field_validator('transaction_id')
    @classmethod
    def validate_transaction_id(cls, v):
        if not re.match(r'^[A-Z]{2}\d{8}[A-Z]{2}$', v):
            raise ValueError('Invalid transaction ID format')
        return v

    @field_validator('timestamp')
    @classmethod
    def validate_timestamp(cls, v):
        if v > datetime.now():
            raise ValueError('Timestamp cannot be in the future')
        return v

These custom validators add extra checks:

• validate_transaction_id: Ensures the ID follows a specific format (2 letters, 8 digits, 2 letters)
• validate_timestamp: Makes sure the transaction time is not in the future

Payment Transaction Model

class PaymentTransaction(TransactionBase):
    payment_method: str
    recipient: str

    @model_validator(mode='after')
    def check_payment_details(self):
        if self.payment_method == 'CASH' and self.amount > 10000:
            raise ValueError('Cash payments cannot exceed 10,000')
        return self

PaymentTransaction extends TransactionBase with:

• payment_method: The method of payment
• recipient: Who receives the payment
• A root validator that checks if cash payments exceed a limit

Refund Transaction Model

class RefundTransaction(TransactionBase):
    reason: str
    original_transaction_id: str

RefundTransaction also extends TransactionBase, adding:

• reason: Why the refund is being made
• original_transaction_id: The ID of the original payment being refunded

Transaction Batch Model

class TransactionBatch(BaseModel):
    batch_id: str
    transactions: List[TransactionBase]
    processed_at: Optional[datetime] = None

    model_config = {
        'populate_by_name': True,
        'json_encoders': {
            datetime: lambda v: v.isoformat(),
            Decimal: lambda v: float(v)
        }
    }

TransactionBatch represents a group of transactions:

• batch_id: An identifier for the batch
• transactions: A list of transactions (can be payments or refunds)
• processed_at: When the batch was processed (optional)

The Config class customizes how Pydantic handles the model:

• Allows populating by field name
• Specifies how to encode datetime and Decimal to JSON

Processing Function

def process_transaction_batch(batch_data: dict):
    try:
        batch = TransactionBatch(**batch_data)
        # Simulate processing
        batch.processed_at = datetime.now()
        return batch.dict(by_alias=True)
    except ValueError as e:
        return f"Validation error: {str(e)}"

This function:

1. Takes a dictionary of batch data
2. Tries to create a TransactionBatch object
3. If successful, it simulates processing and returns the batch as a dictionary
4. If there's a validation error, it returns an error message

Example Usage

batch_data = {
    "batch_id": "BATCH001",
    "transactions": [
        {
            "transaction_id": "TX12345678AB",
            "amount": "100.50",
            "timestamp": "2023-05-15T14:30:00",
            "payment_method": "CARD",
            "recipient": "John Doe"
        },
        {
            "transaction_id": "RX87654321CD",
            "amount": "50.25",
            "timestamp": "2023-05-15T15:45:00",
            "reason": "Product return",
            "original_transaction_id": "TX12345678AB"
        }
    ]
}

result = process_transaction_batch(batch_data)
print(result)

This section demonstrates how to use the models:

1. It creates a dictionary with batch data, including two transactions (a payment and a refund)
2. It calls process_transaction_batch with this data
3. Finally, it prints the result

Running the Example

To run this example:

1. Ensure you have Pydantic installed:

   pip install pydantic
   

2. Run the script:

   python pydantic_transaction_example.py
   

This example shows how Pydantic can handle complex data structures, perform validations, and prepare data for further processing or storage.