You've already forked transaction-tracker
148 lines
4.9 KiB
Markdown
148 lines
4.9 KiB
Markdown
# 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
|
|
|
|
```bash
|
|
# 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
|
|
|
|
```python
|
|
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
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
@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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```bash
|
|
# 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](LICENSE)
|
|
|
|
## Contributing
|
|
|
|
Contributions are welcome! Please feel free to submit a Pull Request. |