from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify from flask_login import login_required from flask_wtf import FlaskForm from wtforms import StringField, SelectField, TextAreaField, SubmitField, DecimalField, DateField, HiddenField from wtforms.validators import DataRequired, Optional, NumberRange from app.extensions import db from app.models.transaction import Transaction from app.models.account import Account from app.models.category import Category from app.services.account_service import calc_balance from datetime import date, datetime from sqlalchemy import or_ transactions_bp = Blueprint('transactions', __name__, url_prefix='/transactions') def _account_choices(): return [(str(a.id), a.name) for a in Account.query.filter_by(is_active=True).order_by(Account.name).all()] def _category_choices(cat_type): cats = Category.query.filter( Category.category_type.in_([cat_type, 'both']), Category.is_active == True, Category.parent_id == None ).order_by(Category.name).all() return [('', '— None —')] + [(str(c.id), c.name) for c in cats] class TransactionForm(FlaskForm): transaction_type = HiddenField(default='expense') account_id = SelectField('Account', validators=[DataRequired()]) category_id = SelectField('Category', validators=[Optional()]) amount = DecimalField('Amount', validators=[DataRequired(), NumberRange(min=0.01)], places=2) description = StringField('Description', validators=[DataRequired()]) date = DateField('Date', validators=[DataRequired()], default=date.today) notes = TextAreaField('Notes', validators=[Optional()]) submit = SubmitField('Save') class TransferForm(FlaskForm): from_account_id = SelectField('From Account', validators=[DataRequired()]) to_account_id = SelectField('To Account', validators=[DataRequired()]) amount = DecimalField('Amount', validators=[DataRequired(), NumberRange(min=0.01)], places=2) description = StringField('Description', default='Transfer') date = DateField('Date', validators=[DataRequired()], default=date.today) notes = TextAreaField('Notes', validators=[Optional()]) submit = SubmitField('Transfer') @transactions_bp.route('/') @login_required def index(): tab = request.args.get('tab', 'expense') # 'income' | 'expense' page = request.args.get('page', 1, type=int) search = request.args.get('q', '').strip() category_id = request.args.get('category_id', '', type=str) account_id = request.args.get('account_id', '', type=str) date_from = request.args.get('date_from', '') date_to = request.args.get('date_to', '') query = Transaction.query.filter( Transaction.transaction_type == tab ).order_by(Transaction.date.desc(), Transaction.id.desc()) if search: query = query.filter(Transaction.description.ilike(f'%{search}%')) if category_id: query = query.filter(Transaction.category_id == int(category_id)) if account_id: query = query.filter(Transaction.account_id == int(account_id)) if date_from: try: query = query.filter(Transaction.date >= datetime.strptime(date_from, '%Y-%m-%d').date()) except ValueError: pass if date_to: try: query = query.filter(Transaction.date <= datetime.strptime(date_to, '%Y-%m-%d').date()) except ValueError: pass pagination = query.paginate(page=page, per_page=30, error_out=False) # Counts for tabs income_count = Transaction.query.filter_by(transaction_type='income').count() expense_count = Transaction.query.filter_by(transaction_type='expense').count() accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all() categories = Category.query.filter( Category.category_type.in_([tab, 'both']), Category.is_active == True ).order_by(Category.name).all() return render_template('transactions/index.html', pagination=pagination, transactions=pagination.items, tab=tab, income_count=income_count, expense_count=expense_count, accounts=accounts, categories=categories, search=search, category_id=category_id, account_id=account_id, date_from=date_from, date_to=date_to) @transactions_bp.route('/new', methods=['GET', 'POST']) @login_required def new(): txn_type = request.args.get('type', 'expense') if txn_type not in ('income', 'expense'): txn_type = 'expense' form = TransactionForm() form.transaction_type.data = txn_type form.account_id.choices = _account_choices() form.category_id.choices = _category_choices(txn_type) if form.validate_on_submit(): txn = Transaction( transaction_type=form.transaction_type.data, account_id=int(form.account_id.data), category_id=int(form.category_id.data) if form.category_id.data else None, amount=form.amount.data, description=form.description.data.strip(), date=form.date.data, notes=form.notes.data, ) db.session.add(txn) db.session.commit() calc_balance(txn.account_id) flash(f'{"Income" if txn_type == "income" else "Expense"} added.', 'success') return redirect(url_for('transactions.index', tab=txn_type)) return render_template('transactions/form.html', form=form, txn_type=txn_type, title=f'New {"Income" if txn_type == "income" else "Expense"}') @transactions_bp.route('//edit', methods=['GET', 'POST']) @login_required def edit(id): txn = db.get_or_404(Transaction, id) form = TransactionForm(obj=txn) form.transaction_type.data = txn.transaction_type form.account_id.choices = _account_choices() form.category_id.choices = _category_choices(txn.transaction_type) # Pre-populate foreign keys as strings for SelectField if request.method == 'GET': form.account_id.data = str(txn.account_id) form.category_id.data = str(txn.category_id) if txn.category_id else '' if form.validate_on_submit(): old_account_id = txn.account_id txn.account_id = int(form.account_id.data) txn.category_id = int(form.category_id.data) if form.category_id.data else None txn.amount = form.amount.data txn.description = form.description.data.strip() txn.date = form.date.data txn.notes = form.notes.data db.session.commit() calc_balance(old_account_id) calc_balance(txn.account_id) flash('Transaction updated.', 'success') return redirect(url_for('transactions.index', tab=txn.transaction_type)) return render_template('transactions/form.html', form=form, txn=txn, txn_type=txn.transaction_type, title='Edit Transaction') @transactions_bp.route('//delete', methods=['POST']) @login_required def delete(id): txn = db.get_or_404(Transaction, id) account_id = txn.account_id txn_type = txn.transaction_type db.session.delete(txn) db.session.commit() calc_balance(account_id) flash('Transaction deleted.', 'info') return redirect(url_for('transactions.index', tab=txn_type)) @transactions_bp.route('/transfer', methods=['GET', 'POST']) @login_required def transfer(): form = TransferForm() form.from_account_id.choices = _account_choices() form.to_account_id.choices = _account_choices() if form.validate_on_submit(): if form.from_account_id.data == form.to_account_id.data: flash('Source and destination accounts must be different.', 'warning') else: txn = Transaction( transaction_type='transfer', account_id=int(form.from_account_id.data), to_account_id=int(form.to_account_id.data), amount=form.amount.data, description=form.description.data.strip() or 'Transfer', date=form.date.data, notes=form.notes.data, ) db.session.add(txn) db.session.commit() calc_balance(txn.account_id) calc_balance(txn.to_account_id) flash('Transfer recorded.', 'success') return redirect(url_for('transactions.index')) return render_template('transactions/transfer.html', form=form)