Transaction Tracker

A lightweight Python library for tracking processed transactions with automatic TTL (Time-To-Live) functionality. This library helps prevent duplicate transaction processing by maintaining a record of previously processed transactions.

Features

• Simple file-based storage: Transactions are stored as individual JSON files on disk
• Automatic TTL: Records automatically expire after a configurable period (default: 3 days)
• Decorator syntax: Easy-to-use decorator for making functions transaction-safe
• No external dependencies: Uses only Python standard library modules
• Customizable: Configure storage location and TTL period to suit your needs

Installation

    # Clone the repository
    git clone https://hub.cybercinch.nz/cybercinch/transaction-tracker.git
    
    # Copy the transaction_tracker.py file to your project
    cp transaction-tracker/transaction_tracker.py /path/to/your/project/

Quick Start

from transaction_tracker import TransactionProcessor, TransactionAlreadyProcessedError

# Create a transaction processor
processor = TransactionProcessor(storage_dir="transaction_records", ttl_days=3)

# Define a transaction processing function
@processor.unique_transaction()
def process_payment(transaction_id, amount, user_id):
    print(f"Processing ${amount} payment for user {user_id}")
    # Your actual payment processing logic here
    return {"status": "success", "amount": amount}

# Use your decorated function
try:
    # This will succeed
    result = process_payment("tx123", 100.00, "user456")
    print(f"Payment result: {result}")
    
    # This will fail (duplicate transaction)
    result = process_payment("tx123", 100.00, "user456")
except TransactionAlreadyProcessedError as e:
    print(f"Expected error: {e}")

Usage

Basic Usage

1. Create a TransactionProcessor instance
2. Decorate your transaction processing functions with @processor.unique_transaction()
3. Handle the TransactionAlreadyProcessedError exception for duplicate transactions

    from transaction_tracker import TransactionProcessor, TransactionAlreadyProcessedError
    import logging
    
    # Configure logging (optional)
    logging.basicConfig(level=logging.INFO)
    
    # Create processor with custom TTL and storage location
    processor = TransactionProcessor(
        storage_dir="/path/to/storage", 
        ttl_days=7  # Keep records for 7 days
    )
    
    # Decorate your function
    @processor.unique_transaction()
    def process_refund(transaction_id, amount, customer_id):
        # Your refund processing logic
        print(f"Processing refund of ${amount} for customer {customer_id}")
        return {"status": "refunded", "amount": amount}
    
    # Use your function and handle potential duplicates
    try:
        result = process_refund("refund_001", 50.00, "customer123")
        print(f"Refund processed: {result}")
    except TransactionAlreadyProcessedError:
        print("This refund was already processed")

Custom Transaction ID Parameter

If your function uses a different parameter name for the transaction ID:

    @processor.unique_transaction(id_arg='refund_id')
    def process_refund(refund_id, amount, customer_id):
        # Function uses refund_id instead of transaction_id
        return {"status": "refunded", "amount": amount}

Manual Cleanup

Records are automatically cleaned up when the processor is initialized, but you can also trigger cleanup manually:

    # Clean up expired transactions
    processor.cleanup()
    
    # Reset/clear all transaction records
    processor.reset()

How It Works

1. When a decorated function is called, the processor checks if the transaction ID has been processed before
2. If the transaction is new, the function executes normally and the transaction is marked as processed
3. If the transaction was already processed, a TransactionAlreadyProcessedError is raised
4. Transaction records automatically expire after the configured TTL period

Configuration Options

Parameter	Default	Description
storage_dir	"transaction_records"	Directory where transaction records are stored
ttl_days	3	Number of days before a transaction record expires

Testing

This library includes a comprehensive test suite using pytest. To run the tests:

    # Install pytest if you don't have it
    pip install pytest
    
    # Run the tests
    pytest transaction_tracker_tests.py -v

Error Handling

The library raises specific exceptions that you can catch and handle:

• TransactionAlreadyProcessedError: Raised when attempting to process a transaction that was already processed
• ValueError: Raised when the transaction ID parameter cannot be found in the function arguments

License

MIT License

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.