persona-finance-tracker-demo/app/api/transactions.py
2025-09-13 17:40:23 +05:30

364 lines
12 KiB
Python

"""
Transactions API Blueprint
This module contains API endpoints for transaction management.
Handles CRUD operations for financial transactions with validation and filtering.
"""
from flask import Blueprint, request, jsonify
from marshmallow import ValidationError
from sqlalchemy.exc import IntegrityError
from datetime import datetime, date
from app.models import Transaction, Category
from app.schemas import (
TransactionCreateSchema, TransactionUpdateSchema,
TransactionListSchema, TransactionFilterSchema
)
from app.extensions import db
# Create transactions blueprint
transactions_bp = Blueprint('transactions', __name__)
# Initialize schemas
transaction_create_schema = TransactionCreateSchema()
transaction_update_schema = TransactionUpdateSchema()
transaction_list_schema = TransactionListSchema(many=True)
transaction_detail_schema = TransactionListSchema()
transaction_filter_schema = TransactionFilterSchema()
@transactions_bp.route('/', methods=['GET'])
def get_transactions():
"""
Get all transactions with optional filtering.
Query Parameters:
- type: string (income/expense) - Filter by transaction type
- category_id: integer - Filter by category ID
- start_date: string (YYYY-MM-DD) - Filter from date
- end_date: string (YYYY-MM-DD) - Filter to date
- page: integer (default: 1) - Page number for pagination
- per_page: integer (default: 20, max: 100) - Items per page
Returns:
- 200: List of transactions with pagination info
"""
try:
# Validate filter parameters
filter_data = transaction_filter_schema.load(request.args.to_dict())
# Start with base query
query = Transaction.query
# Apply filters
if filter_data.get('type'):
query = query.filter(Transaction.type == filter_data['type'])
if filter_data.get('category_id'):
query = query.filter(Transaction.category_id == filter_data['category_id'])
if filter_data.get('start_date'):
query = query.filter(Transaction.transaction_date >= filter_data['start_date'])
if filter_data.get('end_date'):
query = query.filter(Transaction.transaction_date <= filter_data['end_date'])
# Order by transaction_date descending, then by created_at descending
query = query.order_by(Transaction.transaction_date.desc(), Transaction.created_at.desc())
# Pagination
page = filter_data['page']
per_page = filter_data['per_page']
paginated = query.paginate(
page=page,
per_page=per_page,
error_out=False
)
# Serialize transactions with category names
transactions_data = []
for transaction in paginated.items:
data = transaction_detail_schema.dump(transaction)
if transaction.category:
data['category_name'] = transaction.category.name
transactions_data.append(data)
return jsonify({
'success': True,
'data': transactions_data,
'pagination': {
'page': page,
'per_page': per_page,
'total': paginated.total,
'pages': paginated.pages,
'has_next': paginated.has_next,
'has_prev': paginated.has_prev
}
}), 200
except ValidationError as e:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'Filter validation failed',
'details': e.messages
}), 400
except Exception as e:
return jsonify({
'success': False,
'error': 'internal_server_error',
'message': 'Failed to retrieve transactions'
}), 500
@transactions_bp.route('/', methods=['POST'])
def create_transaction():
"""
Create a new transaction.
Request Body:
- description: string (required) - Transaction description
- amount: number (required) - Transaction amount (max 2 decimal places)
- type: string (required) - 'income' or 'expense'
- category_id: integer (optional) - Category ID
- transaction_date: string (optional, default: today) - Date in YYYY-MM-DD format
- notes: string (optional) - Additional notes
Returns:
- 201: Created transaction
- 400: Validation error
- 404: Category not found
"""
try:
# Validate and deserialize request data
json_data = request.get_json()
if not json_data:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'No JSON data provided'
}), 400
validated_data = transaction_create_schema.load(json_data)
# Validate category exists if provided
if validated_data.get('category_id'):
category = Category.query.get(validated_data['category_id'])
if not category:
return jsonify({
'success': False,
'error': 'not_found',
'message': f'Category with ID {validated_data["category_id"]} not found'
}), 404
if not category.is_active:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'Cannot assign transaction to inactive category'
}), 400
# Create new transaction
transaction = Transaction(**validated_data)
db.session.add(transaction)
db.session.commit()
# Return created transaction with category name
data = transaction_detail_schema.dump(transaction)
if transaction.category:
data['category_name'] = transaction.category.name
return jsonify({
'success': True,
'data': data,
'message': 'Transaction created successfully'
}), 201
except ValidationError as e:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'Input validation failed',
'details': e.messages
}), 400
except Exception as e:
db.session.rollback()
return jsonify({
'success': False,
'error': 'internal_server_error',
'message': 'Failed to create transaction'
}), 500
@transactions_bp.route('/<int:transaction_id>', methods=['GET'])
def get_transaction(transaction_id):
"""
Get a specific transaction by ID.
Path Parameters:
- transaction_id: integer - Transaction ID
Returns:
- 200: Transaction details with category information
- 404: Transaction not found
"""
try:
transaction = Transaction.query.get(transaction_id)
if not transaction:
return jsonify({
'success': False,
'error': 'not_found',
'message': f'Transaction with ID {transaction_id} not found'
}), 404
# Serialize transaction with category name
data = transaction_detail_schema.dump(transaction)
if transaction.category:
data['category_name'] = transaction.category.name
return jsonify({
'success': True,
'data': data
}), 200
except Exception as e:
return jsonify({
'success': False,
'error': 'internal_server_error',
'message': 'Failed to retrieve transaction'
}), 500
@transactions_bp.route('/<int:transaction_id>', methods=['PUT'])
def update_transaction(transaction_id):
"""
Update a specific transaction.
Path Parameters:
- transaction_id: integer - Transaction ID
Request Body:
- description: string (optional) - Transaction description
- amount: number (optional) - Transaction amount
- type: string (optional) - 'income' or 'expense'
- category_id: integer (optional) - Category ID
- transaction_date: string (optional) - Date in YYYY-MM-DD format
- notes: string (optional) - Additional notes
Returns:
- 200: Updated transaction
- 400: Validation error
- 404: Transaction or category not found
"""
try:
transaction = Transaction.query.get(transaction_id)
if not transaction:
return jsonify({
'success': False,
'error': 'not_found',
'message': f'Transaction with ID {transaction_id} not found'
}), 404
# Validate and deserialize request data
json_data = request.get_json()
if not json_data:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'No JSON data provided'
}), 400
validated_data = transaction_update_schema.load(json_data)
# Validate category exists if being updated
if 'category_id' in validated_data and validated_data['category_id']:
category = Category.query.get(validated_data['category_id'])
if not category:
return jsonify({
'success': False,
'error': 'not_found',
'message': f'Category with ID {validated_data["category_id"]} not found'
}), 404
if not category.is_active:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'Cannot assign transaction to inactive category'
}), 400
# Update transaction fields
for field, value in validated_data.items():
setattr(transaction, field, value)
db.session.commit()
# Return updated transaction with category name
data = transaction_detail_schema.dump(transaction)
if transaction.category:
data['category_name'] = transaction.category.name
return jsonify({
'success': True,
'data': data,
'message': 'Transaction updated successfully'
}), 200
except ValidationError as e:
return jsonify({
'success': False,
'error': 'validation_failed',
'message': 'Input validation failed',
'details': e.messages
}), 400
except Exception as e:
db.session.rollback()
return jsonify({
'success': False,
'error': 'internal_server_error',
'message': 'Failed to update transaction'
}), 500
@transactions_bp.route('/<int:transaction_id>', methods=['DELETE'])
def delete_transaction(transaction_id):
"""
Delete a specific transaction.
Path Parameters:
- transaction_id: integer - Transaction ID
Returns:
- 200: Transaction deleted successfully
- 404: Transaction not found
"""
try:
transaction = Transaction.query.get(transaction_id)
if not transaction:
return jsonify({
'success': False,
'error': 'not_found',
'message': f'Transaction with ID {transaction_id} not found'
}), 404
# Hard delete the transaction
db.session.delete(transaction)
db.session.commit()
return jsonify({
'success': True,
'message': 'Transaction deleted successfully'
}), 200
except Exception as e:
db.session.rollback()
return jsonify({
'success': False,
'error': 'internal_server_error',
'message': 'Failed to delete transaction'
}), 500