from typing import List, Optional from sqlalchemy.orm import Session from app.crud.user import update_user_balance from app.models.transaction import Transaction, TransactionStatus, TransactionType from app.schemas.transaction import TransactionCreate def get_transaction(db: Session, transaction_id: int) -> Optional[Transaction]: return db.query(Transaction).filter(Transaction.id == transaction_id).first() def get_user_transactions( db: Session, user_id: int, skip: int = 0, limit: int = 100, transaction_type: Optional[TransactionType] = None, ) -> List[Transaction]: query = db.query(Transaction).filter(Transaction.user_id == user_id) if transaction_type: query = query.filter(Transaction.transaction_type == transaction_type) return query.order_by(Transaction.created_at.desc()).offset(skip).limit(limit).all() def create_deposit( db: Session, user_id: int, transaction_in: TransactionCreate, ) -> Transaction: """ Create a deposit transaction and update user balance. """ # Ensure it's a deposit transaction_data = transaction_in.model_dump() transaction_data["transaction_type"] = TransactionType.DEPOSIT transaction_data["status"] = TransactionStatus.COMPLETED transaction_data["user_id"] = user_id # Create transaction db_transaction = Transaction(**transaction_data) db.add(db_transaction) db.commit() db.refresh(db_transaction) # Update user balance update_user_balance(db, user_id, transaction_in.amount) return db_transaction def create_withdrawal( db: Session, user_id: int, transaction_in: TransactionCreate, ) -> Optional[Transaction]: """ Create a withdrawal transaction and update user balance. Returns None if user doesn't have enough balance. """ from app.crud.user import get_user user = get_user(db, user_id) if not user or user.balance < transaction_in.amount: return None # Ensure it's a withdrawal and amount is negative transaction_data = transaction_in.model_dump() transaction_data["transaction_type"] = TransactionType.WITHDRAWAL transaction_data["status"] = TransactionStatus.COMPLETED transaction_data["user_id"] = user_id transaction_data["amount"] = -abs(transaction_in.amount) # Ensure it's negative # Create transaction db_transaction = Transaction(**transaction_data) db.add(db_transaction) db.commit() db.refresh(db_transaction) # Update user balance update_user_balance(db, user_id, transaction_data["amount"]) return db_transaction def update_transaction_status( db: Session, transaction_id: int, status: TransactionStatus, ) -> Optional[Transaction]: """ Update a transaction's status. """ transaction = get_transaction(db, transaction_id) if transaction: transaction.status = status db.add(transaction) db.commit() db.refresh(transaction) return transaction