From 231a9d21934788527e7d18c7879f788a7403b682 Mon Sep 17 00:00:00 2001 From: NguyenND Date: Fri, 5 Jun 2026 18:14:16 -0400 Subject: [PATCH] 06/05 Optimize app --- app/services/account_service.py | 65 ++++++++++++++--------------- app/services/bank_import_service.py | 5 +++ app/services/plaid_service.py | 42 +++++++++---------- app/services/schwab_service.py | 31 +++++++------- app/services/teller_service.py | 51 ++++++++++------------ 5 files changed, 92 insertions(+), 102 deletions(-) diff --git a/app/services/account_service.py b/app/services/account_service.py index 4c13387..63c71ab 100644 --- a/app/services/account_service.py +++ b/app/services/account_service.py @@ -5,48 +5,45 @@ plus any incoming transfers minus outgoing transfers. """ from decimal import Decimal -from sqlalchemy import func +from sqlalchemy import func, case, or_, and_ from app.extensions import db from app.models.account import Account from app.models.transaction import Transaction def calc_balance(account_id): - """Recalculate and persist the balance for a given account.""" - # Income credited to this account - income = db.session.query( - func.coalesce(func.sum(Transaction.amount), 0) + """ + Recalculate and persist the balance for a given account. + Uses a single aggregation query with CASE expressions instead of 4 queries. + """ + row = db.session.query( + func.coalesce(func.sum(case( + (and_(Transaction.account_id == account_id, + Transaction.transaction_type == 'income'), Transaction.amount), + else_=0 + )), 0), + func.coalesce(func.sum(case( + (and_(Transaction.account_id == account_id, + Transaction.transaction_type == 'expense'), Transaction.amount), + else_=0 + )), 0), + func.coalesce(func.sum(case( + (and_(Transaction.account_id == account_id, + Transaction.transaction_type == 'transfer'), Transaction.amount), + else_=0 + )), 0), + func.coalesce(func.sum(case( + (and_(Transaction.to_account_id == account_id, + Transaction.transaction_type == 'transfer'), Transaction.amount), + else_=0 + )), 0), ).filter( - Transaction.account_id == account_id, - Transaction.transaction_type == 'income' - ).scalar() + or_(Transaction.account_id == account_id, + Transaction.to_account_id == account_id) + ).one() - # Expenses debited from this account - expense = db.session.query( - func.coalesce(func.sum(Transaction.amount), 0) - ).filter( - Transaction.account_id == account_id, - Transaction.transaction_type == 'expense' - ).scalar() - - # Transfers out (this account is source) - transfer_out = db.session.query( - func.coalesce(func.sum(Transaction.amount), 0) - ).filter( - Transaction.account_id == account_id, - Transaction.transaction_type == 'transfer' - ).scalar() - - # Transfers in (this account is destination) - transfer_in = db.session.query( - func.coalesce(func.sum(Transaction.amount), 0) - ).filter( - Transaction.to_account_id == account_id, - Transaction.transaction_type == 'transfer' - ).scalar() - - balance = Decimal(str(income)) - Decimal(str(expense)) \ - - Decimal(str(transfer_out)) + Decimal(str(transfer_in)) + income, expense, transfer_out, transfer_in = (Decimal(str(v)) for v in row) + balance = income - expense - transfer_out + transfer_in account = db.session.get(Account, account_id) if account: diff --git a/app/services/bank_import_service.py b/app/services/bank_import_service.py index c7c3663..6188aff 100644 --- a/app/services/bank_import_service.py +++ b/app/services/bank_import_service.py @@ -436,6 +436,11 @@ def _build_cat_id_map(): return {c.name: c.id for c in Category.query.filter_by(is_active=True).all()} +def build_category_map(): + """Public alias — shared by teller, plaid, and schwab services.""" + return _build_cat_id_map() + + # ── Enrichment ──────────────────────────────────────────────────────────────── def _enrich(raw_rows): diff --git a/app/services/plaid_service.py b/app/services/plaid_service.py index a6f2d11..e975108 100644 --- a/app/services/plaid_service.py +++ b/app/services/plaid_service.py @@ -217,12 +217,6 @@ def sync_transactions(item): return added, modified, removed, cursor -def build_category_map(): - from app.models.category import Category - cats = Category.query.filter_by(is_active=True).all() - return {c.name: c.id for c in cats} - - def _map_plaid_category(plaid_cats, txn_type): """Map Plaid category array to a PFM category name.""" if not plaid_cats: @@ -292,6 +286,7 @@ def sync_preview(item): added, _modified, _removed, next_cursor = sync_transactions(item) # Build maps + from app.services.bank_import_service import build_category_map plaid_accounts = PlaidAccount.query.filter_by(item_id=item.id, is_active=True).all() plaid_account_map = {pa.plaid_account_id: pa.pfm_account_id for pa in plaid_accounts} cat_map = build_category_map() @@ -324,9 +319,17 @@ def import_transactions(parsed_txns, next_cursor, item): affected_accounts = set() plaid_account_ids_synced = set() + # Batch duplicate check — one IN query instead of one LIKE per transaction + candidate_notes = {p['notes'] for p in parsed_txns} # 'Plaid:{id}' + existing_notes = { + r[0] for r in + db.session.query(Transaction.notes) + .filter(Transaction.notes.in_(candidate_notes)) + .all() + } if candidate_notes else set() + for p in parsed_txns: - pid = p['plaid_id'] - if Transaction.query.filter(Transaction.notes.like(f'%Plaid:{pid}%')).first(): + if p['notes'] in existing_notes: skipped += 1 continue @@ -344,18 +347,13 @@ def import_transactions(parsed_txns, next_cursor, item): plaid_account_ids_synced.add(p['plaid_account_id']) imported += 1 - db.session.commit() - - # Advance cursor + # Single commit — transactions + cursor + metadata together item.cursor = next_cursor item.last_synced_at = datetime.utcnow() - - # Update last_sync_date per PlaidAccount today = date.today() for pa in PlaidAccount.query.filter_by(item_id=item.id, is_active=True).all(): if pa.plaid_account_id in plaid_account_ids_synced: pa.last_sync_date = today - db.session.commit() # Refresh balances for mapped accounts @@ -441,6 +439,7 @@ def auto_sync_item(item): added, modified, removed, next_cursor = sync_transactions(item) + from app.services.bank_import_service import build_category_map plaid_accounts = PlaidAccount.query.filter_by(item_id=item.id, is_active=True).all() plaid_account_map = {pa.plaid_account_id: pa.pfm_account_id for pa in plaid_accounts} cat_map = build_category_map() @@ -456,17 +455,14 @@ def auto_sync_item(item): imported, skipped = import_transactions(parsed, next_cursor, item) - # Remove transactions that Plaid says are gone (e.g. pending dropped) + # Remove transactions that Plaid says are gone — batch lookup removed_count = 0 if removed: - for r in removed: - tid = r.get('transaction_id', '') - txn = Transaction.query.filter( - Transaction.notes.like(f'%Plaid:{tid}%') - ).first() - if txn: - db.session.delete(txn) - removed_count += 1 + remove_notes = {f'Plaid:{r.get("transaction_id", "")}' for r in removed if r.get('transaction_id')} + txns_to_delete = Transaction.query.filter(Transaction.notes.in_(remove_notes)).all() + for txn in txns_to_delete: + db.session.delete(txn) + removed_count += 1 if removed_count: db.session.commit() log.info('[plaid] auto_sync removed %d transaction(s) for item %s', diff --git a/app/services/schwab_service.py b/app/services/schwab_service.py index 15276e2..7873a09 100644 --- a/app/services/schwab_service.py +++ b/app/services/schwab_service.py @@ -336,12 +336,6 @@ def get_transactions(connection, account_hash, start_date, end_date): return data -def build_category_map(): - from app.models.category import Category - cats = Category.query.filter_by(is_active=True).all() - return {c.name: c.id for c in cats} - - def parse_transaction(schwab_txn, pfm_account_id, cat_id_map): """ Convert a Schwab transaction dict to a PFM-ready dict. @@ -413,6 +407,7 @@ def sync_preview(schwab_account, days_back=90): end_date=today, ) + from app.services.bank_import_service import build_category_map cat_map = build_category_map() return [ parse_transaction(t, schwab_account.pfm_account_id, cat_map) @@ -428,14 +423,20 @@ def import_transactions(parsed_txns, schwab_account): """ from app.extensions import db from app.models.transaction import Transaction - from app.services.account_service import calc_balance imported = skipped = 0 - affected = set() + + # Batch duplicate check — one IN query instead of one LIKE per transaction + candidate_notes = {p['notes'] for p in parsed_txns} # 'Schwab:{id}' + existing_notes = { + r[0] for r in + db.session.query(Transaction.notes) + .filter(Transaction.notes.in_(candidate_notes)) + .all() + } if candidate_notes else set() for p in parsed_txns: - sid = p['schwab_id'] - if Transaction.query.filter(Transaction.notes.like(f'%Schwab:{sid}%')).first(): + if p['notes'] in existing_notes: skipped += 1 continue @@ -448,19 +449,15 @@ def import_transactions(parsed_txns, schwab_account): date = p['date'], notes = p['notes'], )) - if p['account_id']: - affected.add(p['account_id']) imported += 1 - db.session.commit() - + # Single commit — transactions + metadata together schwab_account.last_sync_date = date.today() schwab_account.connection.last_synced_at = datetime.utcnow() if parsed_txns: schwab_account.last_schwab_txn_id = parsed_txns[0]['schwab_id'] db.session.commit() - - for acct_id in affected: - calc_balance(acct_id) + # Balance for Schwab accounts is set by sync_account_snapshot (liquidationValue), + # not by summing transactions — skip calc_balance here. return imported, skipped diff --git a/app/services/teller_service.py b/app/services/teller_service.py index dac91cb..c9e36f5 100644 --- a/app/services/teller_service.py +++ b/app/services/teller_service.py @@ -231,13 +231,6 @@ def parse_transaction(teller_txn, pfm_account_id, category_id_map, is_credit_car } -def build_category_map(): - """Build {pfm_category_name: category_id} from DB.""" - from app.models.category import Category - cats = Category.query.filter_by(is_active=True).all() - return {c.name: c.id for c in cats} - - def sync_preview(teller_account, days_back=90): """ Fetch transactions for a TellerAccount and return a preview list. @@ -272,6 +265,7 @@ def sync_preview(teller_account, days_back=90): ) raise + from app.services.bank_import_service import build_category_map cat_map = build_category_map() is_cc = ( teller_account.account_type == 'credit' or @@ -302,37 +296,38 @@ def import_transactions(parsed_txns, teller_account): skipped = 0 affected_accounts = set() + # Batch duplicate check — one IN query instead of one LIKE per transaction + candidate_notes = {f'Teller:{p["teller_id"]}' for p in parsed_txns} + existing_notes = { + r[0] for r in + db.session.query(Transaction.notes) + .filter(Transaction.notes.in_(candidate_notes)) + .all() + } + for p in parsed_txns: - # Duplicate check: match on teller_id in notes - teller_id = p['teller_id'] - existing = Transaction.query.filter( - Transaction.notes.like(f'%{teller_id}%') - ).first() - if existing: + note_str = f'Teller:{p["teller_id"]}' + if note_str in existing_notes: skipped += 1 continue - txn = Transaction( - account_id=p['account_id'], - category_id=p.get('category_id'), - transaction_type=p['transaction_type'], - amount=p['amount'], - description=p['description'], - date=p['date'], - notes=f"Teller:{teller_id}", - ) - db.session.add(txn) + db.session.add(Transaction( + account_id = p['account_id'], + category_id = p.get('category_id'), + transaction_type = p['transaction_type'], + amount = p['amount'], + description = p['description'], + date = p['date'], + notes = note_str, + )) if p['account_id']: affected_accounts.add(p['account_id']) imported += 1 - db.session.commit() - - # Update sync metadata on both the account and its parent enrollment + # Single commit — transactions + sync metadata together from datetime import datetime - now = datetime.utcnow() teller_account.last_sync_date = date.today() - teller_account.enrollment.last_synced_at = now + teller_account.enrollment.last_synced_at = datetime.utcnow() if parsed_txns: teller_account.last_teller_txn_id = parsed_txns[0]['teller_id'] db.session.commit()