mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-05-19 10:28:48 +02:00
13161b8e49
Some checks failed
Deploy to VPS / deploy (push) Failing after 7s
Backend: FastAPI + PostgreSQL with models for accounts, transactions, and categories. Auto-categorization from merchant patterns, token auth, CRUD endpoints, and seed data for 16 categories and 4 bank accounts. Frontend: Login, Dashboard (account balances + recent charges), Transactions (full CRUD table with search/filter), Cash & Transfers view. Dark theme with emerald/cyan accents, responsive layout. Infrastructure: Updated docker-compose for backend + db services, nginx proxy config for API routing, deploy workflow with secrets. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlmodel import Session, col, select
|
|
|
|
from app.auth import get_current_user
|
|
from app.db import get_session
|
|
from app.models.models import (
|
|
Category,
|
|
Transaction,
|
|
TransactionCreate,
|
|
TransactionRead,
|
|
TransactionSource,
|
|
TransactionUpdate,
|
|
)
|
|
|
|
router = APIRouter(prefix="/transactions", tags=["transactions"])
|
|
|
|
|
|
def auto_categorize(merchant: str, session: Session) -> Optional[int]:
|
|
categories = session.exec(select(Category)).all()
|
|
merchant_lower = merchant.lower()
|
|
for cat in categories:
|
|
if cat.auto_match_patterns:
|
|
patterns = [p.strip().lower() for p in cat.auto_match_patterns.split(",")]
|
|
if any(p in merchant_lower for p in patterns if p):
|
|
return cat.id
|
|
return None
|
|
|
|
|
|
@router.get("/", response_model=list[TransactionRead])
|
|
def list_transactions(
|
|
source: Optional[TransactionSource] = None,
|
|
search: Optional[str] = None,
|
|
category_id: Optional[int] = None,
|
|
limit: int = Query(default=50, le=500),
|
|
offset: int = 0,
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
query = select(Transaction)
|
|
if source:
|
|
query = query.where(Transaction.source == source)
|
|
if category_id:
|
|
query = query.where(Transaction.category_id == category_id)
|
|
if search:
|
|
query = query.where(col(Transaction.merchant).ilike(f"%{search}%"))
|
|
query = query.order_by(col(Transaction.date).desc()).offset(offset).limit(limit)
|
|
return session.exec(query).all()
|
|
|
|
|
|
@router.get("/recent", response_model=list[TransactionRead])
|
|
def recent_transactions(
|
|
limit: int = Query(default=5, le=20),
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
query = (
|
|
select(Transaction)
|
|
.where(Transaction.source == TransactionSource.CREDIT_CARD)
|
|
.order_by(col(Transaction.date).desc())
|
|
.limit(limit)
|
|
)
|
|
return session.exec(query).all()
|
|
|
|
|
|
@router.post("/", response_model=TransactionRead, status_code=201)
|
|
def create_transaction(
|
|
data: TransactionCreate,
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
tx = Transaction.model_validate(data)
|
|
if tx.category_id is None:
|
|
tx.category_id = auto_categorize(tx.merchant, session)
|
|
session.add(tx)
|
|
session.commit()
|
|
session.refresh(tx)
|
|
return tx
|
|
|
|
|
|
@router.patch("/{transaction_id}", response_model=TransactionRead)
|
|
def update_transaction(
|
|
transaction_id: int,
|
|
data: TransactionUpdate,
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
tx = session.get(Transaction, transaction_id)
|
|
if not tx:
|
|
raise HTTPException(status_code=404, detail="Transaction not found")
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(tx, key, value)
|
|
session.add(tx)
|
|
session.commit()
|
|
session.refresh(tx)
|
|
return tx
|
|
|
|
|
|
@router.delete("/{transaction_id}", status_code=204)
|
|
def delete_transaction(
|
|
transaction_id: int,
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
tx = session.get(Transaction, transaction_id)
|
|
if not tx:
|
|
raise HTTPException(status_code=404, detail="Transaction not found")
|
|
session.delete(tx)
|
|
session.commit()
|