Index: ckend-python/auth.py =================================================================== --- backend-python/auth.py (revision 0ba4abde554eb4472ff4126fe101af6200437345) +++ (revision ) @@ -1,46 +1,0 @@ -import jwt -from datetime import datetime, timedelta -from fastapi import HTTPException, status -from passlib.hash import bcrypt - -def hash_password(password: str) -> str: - return bcrypt.hash(password) - -def verify_password(password: str, hashed_password: str) -> bool: - return bcrypt.verify(password, hashed_password) - -def is_admin(email: str) -> bool: - return email in ["admin@fein.com"] - -def create_access_token(data: dict): - to_encode = data.copy() - expire = datetime.utcnow() + timedelta(minutes= 30) - to_encode.update({"exp": expire}) - - # Ensure `sub` is a string - if "sub" in to_encode and not isinstance(to_encode["sub"], str): - to_encode["sub"] = str(to_encode["sub"]) - - # Add other claims (e.g., email) - if "email" in to_encode and not isinstance(to_encode["email"], str): - raise ValueError("Email must be a string") - - encoded_jwt = jwt.encode(to_encode, "A1B2C3D4E5F6G7H8I9J0K", algorithm="HS256") - return encoded_jwt - -def decode_access_token(token: str): - try: - payload = jwt.decode(token, "A1B2C3D4E5F6G7H8I9J0K", algorithms=["HS256"]) - return payload - except jwt.ExpiredSignatureError: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Token has expired", - headers={"WWW-Authenticate": "Bearer"}, - ) - except jwt.PyJWTError: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid token", - headers={"WWW-Authenticate": "Bearer"}, - ) Index: ckend-python/database.py =================================================================== --- backend-python/database.py (revision 0ba4abde554eb4472ff4126fe101af6200437345) +++ (revision ) @@ -1,34 +1,0 @@ -import os -from dotenv import load_dotenv -from sqlalchemy import create_engine -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy.orm import sessionmaker - -# Load environment variables -load_dotenv() - -# Database URL -DATABASE_URL = os.getenv("DATABASE_URL") - -# Create the database engine -engine = create_engine(DATABASE_URL, pool_pre_ping=True) - -# Create a configured "Session" class -SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) - -# Base class for models -Base = declarative_base() - -try: - with engine.connect() as connection: - print("Database connection successful") -except Exception as e: - print(f"Failed to connect to the database: {e}") - -# Dependency to get the database session -def get_db(): - db = SessionLocal() - try: - yield db - finally: - db.close() Index: ckend-python/main.py =================================================================== --- backend-python/main.py (revision 0ba4abde554eb4472ff4126fe101af6200437345) +++ (revision ) @@ -1,1095 +1,0 @@ -from fastapi import FastAPI, Depends, HTTPException -from sqlalchemy.orm import Session -from app.database import get_db -from app.models import User, TransactionAccount, Transaction, TransactionBreakdown, Tag, TagAssignedToTransaction -from pydantic import BaseModel -from typing import List, Optional -from datetime import datetime -from fastapi.security import OAuth2PasswordBearer -from app.auth import create_access_token, decode_access_token, is_admin, hash_password, verify_password -from sqlalchemy import func, literal_column, select - -# Initialize FastAPI app -app = FastAPI() - -oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login/") - -# Pydantic schemas for validation and response models -class UserCreate(BaseModel): - user_name: str - email: str - password: str - -class UserResponse(BaseModel): - user_id: int - user_name: str - email: str - - class Config: - from_attributes = True - -class TransactionAccountCreate(BaseModel): - account_name: str - balance: float - -class TransactionAccountResponse(BaseModel): - transaction_account_id: int - account_name: str - balance: float - - class Config: - from_attributes = True - -class TransactionBreakdownResponse(BaseModel): - transaction_account_id: int - earned_amount: float - spent_amount: float - - class Config: - from_attributes = True - -class TransactionCreateRequest(BaseModel): - transaction_name: str - amount: float = 0.0 # Default to 0 if not provided - date: Optional[datetime] = None # Default to None, will use current time if not provided - tag_id: Optional[int] = None # Optional tag - target_account_id: int # Mandatory target account - breakdowns: Optional[List[TransactionBreakdownResponse]] = None # Optional list of breakdowns - - -class TransactionUpdate(BaseModel): - transaction_name: str = None - amount: float = None - net_amount: float = None - date: datetime = None - -class TransactionResponse(BaseModel): - transaction_id: int - transaction_name: str - amount: float - net_amount: float - date: datetime - - class Config: - json_encoders = { - datetime: lambda v: v.isoformat() # Serialize datetime as ISO 8601 string - } - from_attributes = True - -class TagCreate(BaseModel): - tag_name: str - -class TagResponse(BaseModel): - tag_id: int - tag_name: str - - class Config: - from_attributes = True - -class TagAssign(BaseModel): - transaction_id: int - tag_id: int - -class AuthRequest(BaseModel): - user_name: Optional[str] = None - email: str - password: str - -class AuthResponse(BaseModel): - access_token: str - token_type: str - -# Dependency to get the current user -def get_current_user( - token: str = Depends(oauth2_scheme), - db: Session = Depends(get_db) -): - """ - Retrieves the current user based on the access token. - """ - try: - payload = decode_access_token(token) - user_id = payload.get("sub") - if not user_id: - raise HTTPException( - status_code=401, - detail="Invalid authentication credentials", - headers={"WWW-Authenticate": "Bearer"} - ) - - user = db.query(User).filter(User.user_id == user_id).first() - if not user: - raise HTTPException( - status_code=401, - detail="Invalid authentication credentials", - headers={"WWW-Authenticate": "Bearer"} - ) - - return user - except Exception as e: - print(f"Error decoding token or fetching user: {e}") - raise HTTPException( - status_code=401, - detail="Invalid authentication credentials", - headers={"WWW-Authenticate": "Bearer"} - ) - -# Routes -@app.get("/") -def read_root(): - return {"message": "Welcome to the Fein Prototype API"} - -@app.post("/auth/register/", response_model=AuthResponse) -def register( - auth_request: AuthRequest, - db: Session = Depends(get_db) -): - # Check if email already exists - existing_user = db.query(User).filter(User.email == auth_request.email).first() - if existing_user: - raise HTTPException( - status_code=400, - detail="Email already registered" - ) - - # Hash password and create new user - hashed_password = hash_password(auth_request.password) - new_user = User( - user_name=auth_request.user_name, - email=auth_request.email, - password=hashed_password - ) - db.add(new_user) - db.commit() - db.refresh(new_user) - - # Return access token - access_token = create_access_token({"sub": new_user.user_id, "email": new_user.email}) - return {"access_token": access_token, "token_type": "bearer"} - -@app.post("/auth/login/", response_model=AuthResponse) -def login( - auth_request: AuthRequest, - db: Session = Depends(get_db) -): - # Verify email and password - user = db.query(User).filter(User.email == auth_request.email).first() - if not user or not verify_password(auth_request.password, user.password): - raise HTTPException( - status_code=401, - detail="Invalid email or password" - ) - - # Return access token - access_token = create_access_token({"sub": user.user_id, "email": user.email}) - return {"access_token": access_token, "token_type": "bearer"} - -@app.get("/admin/accounts/", response_model=List[TransactionAccountResponse]) -def admin_get_all_accounts( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Admin can fetch all transaction accounts. - """ - if not is_admin(user.email): - raise HTTPException( - status_code=403, - detail="Access denied" - ) - return db.query(TransactionAccount).all() - - -@app.post("/accounts/", response_model=TransactionAccountResponse) -def create_account( - account: TransactionAccountCreate, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - new_account = TransactionAccount( - account_name=account.account_name, - balance=account.balance, - user_id=user.user_id - ) - db.add(new_account) - db.commit() - db.refresh(new_account) - return new_account - -@app.get("/accounts/", response_model=List[TransactionAccountResponse]) -def get_accounts( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Admin can fetch all accounts, regular users only their accounts. - """ - query = db.query(TransactionAccount) - - if is_admin(user.email): - return query.all() - - return query.filter(TransactionAccount.user_id == user.user_id).all() - - -@app.post("/transactions/", response_model=TransactionResponse) -def create_transaction( - transaction_request: TransactionCreateRequest, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Admins can create transactions for any account; regular users only for their accounts. - Create a transaction and associate it with the user's accounts via breakdowns. - """ - # Admin bypasses ownership checks - if not is_admin(user.email): - # Validate target account ownership - target_account = db.query(TransactionAccount).filter( - TransactionAccount.transaction_account_id == transaction_request.target_account_id, - TransactionAccount.user_id == user.user_id - ).first() - if not target_account: - raise HTTPException( - status_code=403, - detail="Access denied to target account." - ) - - # Create transaction - new_transaction = Transaction( - transaction_name=transaction_request.transaction_name, - amount=transaction_request.amount, - net_amount=0.0, # Will be updated based on breakdowns - date=transaction_request.date or datetime.utcnow(), # Use current UTC time if not provided - ) - db.add(new_transaction) - db.commit() - db.refresh(new_transaction) - - # Associate a tag, if provided - if transaction_request.tag_id: - tag = db.query(Tag).filter(Tag.tag_id == transaction_request.tag_id).first() - if tag: - tag_assignment = TagAssignedToTransaction( - transaction_id=new_transaction.transaction_id, - tag_id=tag.tag_id - ) - db.add(tag_assignment) - - # Add breakdowns - net_amount = 0.0 - if transaction_request.breakdowns: - for breakdown in transaction_request.breakdowns: - # Validate breakdown account ownership - breakdown_account = db.query(TransactionAccount).filter( - TransactionAccount.transaction_account_id == breakdown.transaction_account_id, - TransactionAccount.user_id == user.user_id - ).first() - if not breakdown_account: - raise HTTPException( - status_code=403, - detail=f"Access denied to breakdown account {breakdown.transaction_account_id}." - ) - - # Create breakdown - new_breakdown = TransactionBreakdown( - transaction_id=new_transaction.transaction_id, - transaction_account_id=breakdown.transaction_account_id, - earned_amount=breakdown.earned_amount, - spent_amount=breakdown.spent_amount - ) - db.add(new_breakdown) - - # Calculate net amount - net_amount += breakdown.earned_amount - breakdown.spent_amount - - # Update transaction's net amount - new_transaction.net_amount = net_amount - db.commit() - db.refresh(new_transaction) - - return new_transaction - -@app.get("/transactions/", response_model=List[TransactionResponse]) -def get_transactions( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Fetch transactions based on user role, excluding placeholder transactions. - - Admin: Fetch all non-placeholder transactions. - - Regular User: Fetch non-placeholder transactions tied to the user's accounts via transaction breakdowns. - """ - query = db.query(Transaction) - - if is_admin(user.email): - # Admins see all non-placeholder transactions - transactions = ( - query - .filter(~Transaction.transaction_name.like("Tag_%_placeholder")) - .all() - ) - else: - transactions = ( - query - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(TransactionAccount.user_id == user.user_id) - .filter(~Transaction.transaction_name.like("Tag_%_placeholder")) # Exclude placeholders - .all() - ) - - return [ - { - **transaction.__dict__, - "date": transaction.date.isoformat() # Convert datetime to ISO 8601 string - } - for transaction in transactions - ] - -@app.get("/transactions/{transaction_id}", response_model=TransactionResponse) -def get_transaction_by_id( - transaction_id: int, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Retrieve a single transaction by its ID, ensuring access is restricted - to the transaction creator or an admin user. - """ - # If the user is an admin, they can access any transaction - if is_admin(user.email): - transaction = db.query(Transaction).filter(Transaction.transaction_id == transaction_id).first() - if not transaction: - raise HTTPException( - status_code=404, - detail="Transaction not found." - ) - return transaction - - # Otherwise, restrict access to the transaction creator - transaction = ( - db.query(Transaction) - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(Transaction.transaction_id == transaction_id) - .filter(TransactionAccount.user_id == user.user_id) - .first() - ) - if not transaction: - raise HTTPException( - status_code=404, - detail="Transaction not found or access denied." - ) - - return transaction - -@app.get("/transactions/{transaction_id}/breakdowns", response_model=List[TransactionBreakdownResponse]) -def get_transaction_breakdowns( - transaction_id: int, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Fetch transaction breakdowns for a specific transaction. - """ - breakdowns = ( - db.query(TransactionBreakdown) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(TransactionBreakdown.transaction_id == transaction_id) - .filter(TransactionAccount.user_id == user.user_id) - .all() - ) - - if not breakdowns: - raise HTTPException( - status_code=404, - detail="No breakdowns found for this transaction." - ) - - return breakdowns - - -@app.put("/transactions/{transaction_id}", response_model=TransactionResponse) -def update_transaction( - transaction_id: int, - transaction_update: TransactionUpdate, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Admins can update any transaction - Regular users update a transaction only if it belongs to the logged-in user. - """ - query = db.query(Transaction) - - if is_admin(user.email): - transaction = ( - query - .filter(Transaction.transaction_id == transaction_id) - .first() - ) - if not transaction: - raise HTTPException( - status_code=404, - detail="Transaction not found." - ) - else: - transaction = ( - query - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(Transaction.transaction_id == transaction_id) - .filter(TransactionAccount.user_id == user.user_id) - .first() - ) - if not transaction: - raise HTTPException( - status_code=404, - detail="Transaction not found or access denied." - ) - - # Update transaction fields - for key, value in transaction_update.dict(exclude_unset=True).items(): - setattr(transaction, key, value) - - db.commit() - db.refresh(transaction) - return transaction - -@app.delete("/transactions/{transaction_id}") -def delete_transaction( - transaction_id: int, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Admins can delete any transaction - Regular users can delete a transaction only if it belongs to the logged-in user. - """ - query = db.query(Transaction) - - if is_admin(user.email): - transaction = ( - query - .filter(Transaction.transaction_id == transaction_id) - .first() - ) - if not transaction: - raise HTTPException( - status_code=404, - detail="Transaction not found." - ) - else: - transaction = ( - query - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(Transaction.transaction_id == transaction_id) - .filter(TransactionAccount.user_id == user.user_id) - .first() - ) - if not transaction: - raise HTTPException( - status_code=404, - detail="Transaction not found or access denied." - ) - - db.delete(transaction) - db.commit() - return {"message": "Transaction deleted successfully"} - - -@app.post("/tags/", response_model=TagResponse) -def create_tag( - tag: TagCreate, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Create a tag associated with the logged-in user by linking it to a placeholder transaction. - """ - # Create the tag - new_tag = Tag(tag_name=tag.tag_name) - db.add(new_tag) - db.commit() - db.refresh(new_tag) - - # Create a dummy transaction linked to the user's first account - user_account = ( - db.query(TransactionAccount) - .filter(TransactionAccount.user_id == user.user_id) - .first() - ) - if not user_account: - raise HTTPException( - status_code=403, - detail="No account available to associate with the tag." - ) - - # Associate the tag with a dummy transaction for the user - dummy_transaction = Transaction( - transaction_name=f"Tag_{new_tag.tag_id}_placeholder", - amount=0, - net_amount=0, - date=datetime.utcnow(), - ) - db.add(dummy_transaction) - db.commit() - db.refresh(dummy_transaction) - - # Link the dummy transaction to the user's account - dummy_breakdown = TransactionBreakdown( - transaction_id=dummy_transaction.transaction_id, - transaction_account_id=user_account.transaction_account_id, - earned_amount=0, - spent_amount=0, - ) - db.add(dummy_breakdown) - - # Associate the tag with the dummy transaction - tag_assignment = TagAssignedToTransaction( - transaction_id=dummy_transaction.transaction_id, - tag_id=new_tag.tag_id, - ) - db.add(tag_assignment) - db.commit() - - return new_tag - -@app.get("/tags/", response_model=List[TagResponse]) -def get_tags( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Admins can fetch all tags - Regular users can retrieve tags accessible to the logged-in user based on their transactions. - """ - if is_admin(user.email): - return db.query(Tag).all() - - accessible_tags = ( - db.query(Tag) - .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id) - .join(Transaction, TagAssignedToTransaction.transaction_id == Transaction.transaction_id) - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(TransactionAccount.user_id == user.user_id) - .distinct() - .all() - ) - return accessible_tags - -@app.post("/tags/assign/", response_model=dict) -def assign_tag_to_transaction( - tag_assign: TagAssign, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Assign a tag to a transaction. - - Admins can assign any tag to any transaction. - - Regular users can assign a tag if: - - The transaction belongs to them. - - The tag is accessible (created by them or linked to their transactions). - """ - # Ensure the transaction belongs to the logged-in user - transaction = ( - db.query(Transaction) - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(Transaction.transaction_id == tag_assign.transaction_id) - .filter(TransactionAccount.user_id == user.user_id) - .first() - ) - if not transaction: - raise HTTPException(status_code=404, detail="Transaction not found or access denied.") - - # Ensure the tag is accessible to the logged-in user - tag_accessible = ( - db.query(Tag) - .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id, isouter=True) - .join(Transaction, TagAssignedToTransaction.transaction_id == Transaction.transaction_id, isouter=True) - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id, isouter=True) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id, isouter=True) - .filter(Tag.tag_id == tag_assign.tag_id) - .filter( - (TransactionAccount.user_id == user.user_id) | # Tag linked to the user's transactions - (TransactionAccount.user_id.is_(None)) # Newly created tag not yet assigned - ) - .first() - ) - if not tag_accessible: - raise HTTPException( - status_code=404, - detail="Access denied to the tag." - ) - - # Check if the tag is already assigned to the transaction - existing_assignment = ( - db.query(TagAssignedToTransaction) - .filter( - TagAssignedToTransaction.transaction_id == tag_assign.transaction_id, - TagAssignedToTransaction.tag_id == tag_assign.tag_id, - ) - .first() - ) - if existing_assignment: - raise HTTPException( - status_code=400, - detail="Tag already assigned to this transaction." - ) - - # Assign the tag to the transaction - assignment = TagAssignedToTransaction( - transaction_id=tag_assign.transaction_id, - tag_id=tag_assign.tag_id, - ) - db.add(assignment) - db.commit() - - return {"message": "Tag assigned to transaction successfully"} - -@app.get("/tags/transaction/{transaction_id}", response_model=List[TagResponse]) -def get_transaction_tags_for_user( - transaction_id: int, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Retrieve tags for a specific transaction. - - Admins can retrieve tags for any transaction. - - Regular users can retrieve tags if the transaction belongs to them. - """ - # Admins can access tags for any transaction - if is_admin(user.email): - tags = ( - db.query(Tag) - .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id) - .filter(TagAssignedToTransaction.transaction_id == transaction_id) - .all() - ) - return tags - - # Check if the transaction belongs to the user - transaction = ( - db.query(Transaction) - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(Transaction.transaction_id == transaction_id) - .filter(TransactionAccount.user_id == user.user_id) - .first() - ) - if not transaction: - raise HTTPException( - status_code=403, - detail="Access denied" - ) - - # Retrieve tags for the transaction - tags = ( - db.query(Tag) - .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id) - .filter(TagAssignedToTransaction.transaction_id == transaction_id) - .all() - ) - - return tags - -@app.get("/reports/total-spending", response_model=dict) -def get_total_spending( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Calculate and return total spending for the logged-in user. - - Admins can view total spending for all users. - """ - try: - query = db.query( - func - .sum(Transaction.amount) - .label("total_spent") - ) - - if is_admin(user.email): - # Admin: Total spending for all users - total_spent = ( - query - .filter(Transaction.amount > 0) - .scalar() - ) - else: - # Regular User: Total spending for their accounts - total_spent = ( - query - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter(TransactionAccount.user_id == user.user_id) - .filter(Transaction.amount > 0) - .scalar() - ) - - return {"total_spent": total_spent or 0.0} - except Exception as e: - print(f"Error calculating total spending: {e}") - raise HTTPException( - status_code=500, - detail="Failed to calculate total spending." - ) - -@app.get("/reports/spending-by-category", response_model=dict) -def get_spending_by_category( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Calculate and return spending grouped by category (tags) for the logged-in user. - - Admins can view spending by category for all users. - """ - try: - # Base query - query = db.query( - Tag.tag_name, - func.sum(Transaction.amount).label("total_spent") - ).join( - TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id - ).join( - Transaction, TagAssignedToTransaction.transaction_id == Transaction.transaction_id - ).filter( - Transaction.amount > 0 # Include only positive amounts - ) - - # Apply filters for regular users - if not is_admin(user.email): - query = query.join( - TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id - ).join( - TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id - ).filter( - TransactionAccount.user_id == user.user_id - ) - - # Group by tag and calculate the total spending for each category - spending_by_category = query.group_by(Tag.tag_name).all() - - # Prepare the response as a dictionary - response = {row.tag_name: float(row.total_spent or 0) for row in spending_by_category} - return {"spending_by_category": response} - - except Exception as e: - print(f"Error calculating spending by category: {e}") - raise HTTPException( - status_code=500, - detail="Failed to calculate spending by category." - ) - -@app.get("/reports/spending-by-date-range", response_model=dict) -def get_spending_by_date_range( - start_date: str, # Expecting date in 'YYYY-MM-DD' format - end_date: str, - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Calculate and return spending within a specified date range for the logged-in user. - - Admins can view spending within the date range for all users. - """ - try: - # Convert input dates to `datetime` - start_date_parsed = datetime.strptime(start_date, "%Y-%m-%d") - end_date_parsed = datetime.strptime(end_date, "%Y-%m-%d") - - # Query base - query = db.query(func.sum(Transaction.amount).label("total_spent")) - - if is_admin(user.email): - # Admin: Total spending for all users within the date range - total_spent = ( - query - .filter( - Transaction.date >= start_date_parsed, - Transaction.date <= end_date_parsed, - Transaction.amount > 0 - ) - .scalar() - ) - else: - # Regular User: Total spending within the date range for their accounts - total_spent = ( - query - .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .filter( - TransactionAccount.user_id == user.user_id, - Transaction.date >= start_date_parsed, - Transaction.date <= end_date_parsed, - Transaction.amount > 0 - ) - .scalar() - ) - - # Return result - return {"total_spent": total_spent or 0.0} - except Exception as e: - print(f"Error calculating spending by date range: {e}") - raise HTTPException( - status_code=500, - detail="Failed to calculate spending by date range." - ) - -@app.get("/reports/exceeding-transactions", response_model=List[dict]) -def get_exceeding_transactions( - account_name: Optional[str] = None, # Allow filtering by account name - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Retrieve a list of transactions that exceeded the balance of an account, sorted chronologically. - - Admins can view for all users. - - Regular users can view for their own accounts. - """ - # Define the subquery to calculate `calculated_balance` using a window function - subquery = ( - db.query( - Transaction.transaction_id, - Transaction.transaction_name, - Transaction.date.label("transaction_date"), - TransactionAccount.account_name, - User.user_id, - User.user_name, - TransactionBreakdown.spent_amount.label("transaction_amount"), - func.sum(TransactionBreakdown.earned_amount - TransactionBreakdown.spent_amount) - .over( - partition_by=TransactionBreakdown.transaction_account_id, - order_by=Transaction.date - ) - .label("calculated_balance"), - ) - .join(TransactionAccount, TransactionAccount.transaction_account_id == TransactionBreakdown.transaction_account_id) - .join(User, TransactionAccount.user_id == User.user_id) - .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .subquery() - ) - - query = db.query( - subquery.c.transaction_id, - subquery.c.transaction_name, - subquery.c.transaction_date, - subquery.c.account_name, - subquery.c.user_id, - subquery.c.user_name, - subquery.c.transaction_amount, - subquery.c.calculated_balance, - ).filter( - subquery.c.transaction_amount > subquery.c.calculated_balance, # Filter where transaction amount exceeds balance - subquery.c.transaction_amount > 0, # Filter for positive transactions - ) - - if account_name: - query = query.filter(subquery.c.account_name == account_name) - - # Apply user-specific filtering for non-admins - if not is_admin(user.email): - query = query.filter(subquery.c.user_id == user.user_id) - - # Order results - query = query.order_by( - subquery.c.user_id, - subquery.c.account_name, - subquery.c.transaction_date.desc(), - ) - - # Execute the query and fetch results - results = query.all() - - if not results: - return [] - - # Prepare response - response = [ - { - "user_id": row.user_id, - "user_name": row.user_name, - "account_name": row.account_name, - "transaction_id": row.transaction_id, - "transaction_name": row.transaction_name, - "transaction_amount": row.transaction_amount, - "transaction_date": row.transaction_date, - "calculated_balance": row.calculated_balance, - } - for row in results - ] - - return response - -@app.get("/reports/exceeding-current-balance", response_model=List[dict]) -def get_exceeding_current_balance( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Retrieve a list of transactions that exceed the current balance of accounts. - - Admins can view for all users. - - Regular users can view for their own accounts. - """ - query = ( - db.query( - User.user_id, - User.user_name, - TransactionAccount.account_name, - TransactionAccount.balance.label("current_balance"), - Transaction.transaction_id, - Transaction.transaction_name, - TransactionBreakdown.spent_amount.label("transaction_amount"), - Transaction.date.label("transaction_date"), - ) - .join(TransactionAccount, TransactionAccount.user_id == User.user_id) - .join(TransactionBreakdown, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .filter(TransactionBreakdown.spent_amount > TransactionAccount.balance) # Transactions exceeding account balance - .filter(TransactionBreakdown.spent_amount > 0) # Positive transactions only - ) - - # Apply user-specific filtering for non-admins - if not is_admin(user.email): - query = query.filter(TransactionAccount.user_id == user.user_id) - - # Order results - results = query.order_by(User.user_id, TransactionAccount.account_name, Transaction.date.desc()).all() - - # Prepare response - return [ - { - "user_id": row.user_id, - "user_name": row.user_name, - "account_name": row.account_name, - "current_balance": float(row.current_balance), - "transaction_id": row.transaction_id, - "transaction_name": row.transaction_name, - "transaction_amount": float(row.transaction_amount), - "transaction_date": row.transaction_date, - } - for row in results - ] - -@app.get("/reports/exceeding-total-balances", response_model=List[dict]) -def get_exceeding_total_balances( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Retrieve a chronological list of transactions that exceed the calculated total balances for all accounts. - - Admins can view for all users. - - Regular users can view for their own accounts. - """ - # Subquery to calculate the running balance using a window function - subquery = ( - db.query( - User.user_id.label("user_id"), - User.user_name.label("user_name"), - Transaction.transaction_id.label("transaction_id"), - Transaction.transaction_name.label("transaction_name"), - Transaction.date.label("transaction_date"), - TransactionBreakdown.spent_amount.label("transaction_amount"), - func.sum(TransactionBreakdown.earned_amount - TransactionBreakdown.spent_amount) - .over(partition_by=User.user_id, order_by=Transaction.date) - .label("calculated_total_balance") - ) - .join(TransactionAccount, TransactionAccount.user_id == User.user_id) - .join(TransactionBreakdown, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .filter(TransactionBreakdown.spent_amount > 0) # Only positive transactions - .subquery() - ) - - # Outer query to filter transactions exceeding the calculated balance - query = db.query( - subquery.c.user_id, - subquery.c.user_name, - subquery.c.transaction_id, - subquery.c.transaction_name, - subquery.c.transaction_date, - subquery.c.transaction_amount, - subquery.c.calculated_total_balance - ).filter( - subquery.c.transaction_amount > subquery.c.calculated_total_balance # Exceeds the total balance - ) - - # Apply user-specific filtering for non-admin users - if not is_admin(user.email): - query = query.filter(subquery.c.user_id == user.user_id) - - # Execute the query and return results - results = query.order_by(subquery.c.user_id, subquery.c.transaction_date.desc()).all() - - return [ - { - "user_id": row.user_id, - "user_name": row.user_name, - "transaction_id": row.transaction_id, - "transaction_name": row.transaction_name, - "transaction_date": row.transaction_date, - "transaction_amount": float(row.transaction_amount), - "calculated_total_balance": float(row.calculated_total_balance), - } - for row in results - ] - -@app.get("/reports/exceeding-user-total-balance", response_model=List[dict]) -def get_exceeding_user_total_balance( - user: User = Depends(get_current_user), - db: Session = Depends(get_db) -): - """ - Retrieve a list of users whose transactions exceed the total balance of all their accounts. - - Admins can view results for all users. - - Regular users can only see their own data. - """ - # Subquery to calculate the total balance for each user - total_balance_subquery = ( - db.query( - TransactionAccount.user_id.label("user_id"), - func.sum(TransactionAccount.balance).label("total_balance") - ) - .group_by(TransactionAccount.user_id) - .subquery() - ) - - # Main query - query = ( - db.query( - User.user_id, - User.user_name, - func.sum(TransactionBreakdown.spent_amount).label("total_transaction_amount"), - total_balance_subquery.c.total_balance.label("user_total_balance") - ) - .join(TransactionAccount, TransactionAccount.user_id == User.user_id) - .join(TransactionBreakdown, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) - .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) - .join(total_balance_subquery, total_balance_subquery.c.user_id == User.user_id) - .filter(Transaction.date <= func.current_date()) # Only transactions up to the current date - .group_by(User.user_id, User.user_name, total_balance_subquery.c.total_balance) - .having(func.sum(TransactionBreakdown.spent_amount) > total_balance_subquery.c.total_balance) # Exceeds total balance - .order_by(User.user_id) - ) - - # Apply user-specific filtering for non-admins - if not is_admin(user.email): - query = query.filter(User.user_id == user.user_id) - - # Execute the query - results = query.all() - - # Prepare response - return [ - { - "user_id": row.user_id, - "user_name": row.user_name, - "total_transaction_amount": float(row.total_transaction_amount), - "user_total_balance": float(row.user_total_balance), - } - for row in results - ] - Index: ckend-python/models.py =================================================================== --- backend-python/models.py (revision 0ba4abde554eb4472ff4126fe101af6200437345) +++ (revision ) @@ -1,57 +1,0 @@ -from sqlalchemy import Column, Integer, String, Numeric, ForeignKey, DateTime, Boolean -from sqlalchemy.orm import relationship -from sqlalchemy.ext.declarative import declarative_base - -Base = declarative_base() - -# Define Models -class User(Base): - __tablename__ = 'user' - user_id = Column(Integer, primary_key=True, autoincrement=True) - user_name = Column(String(50), nullable=False) - email = Column(String(100), unique=True, nullable=False) - password = Column(String(255), nullable=False) - accounts = relationship("TransactionAccount", back_populates="user") - -class TransactionAccount(Base): - __tablename__ = 'transaction_account' - transaction_account_id = Column(Integer, primary_key=True, autoincrement=True) - account_name = Column(String(50), nullable=False) - balance = Column(Numeric(10, 2), default=0, nullable=False) - user_id = Column(Integer, ForeignKey('user.user_id')) - user = relationship("User", back_populates="accounts") - -class Tag(Base): - __tablename__ = 'tag' - tag_id = Column(Integer, primary_key=True, autoincrement=True) - tag_name = Column(String(50), nullable=False) - transactions = relationship( - "Transaction", secondary="tag_assigned_to_transaction", back_populates="tags" - ) - -class Transaction(Base): - __tablename__ = 'transaction' - transaction_id = Column(Integer, primary_key=True, autoincrement=True) - transaction_name = Column(String(100), nullable=False) - amount = Column(Numeric(10, 2), nullable=False) - net_amount = Column(Numeric(10, 2), nullable=False) - date = Column(DateTime(timezone=True), nullable=False) - breakdowns = relationship("TransactionBreakdown", back_populates="transaction") - tags = relationship( - "Tag", secondary="tag_assigned_to_transaction", back_populates="transactions" - ) - -class TransactionBreakdown(Base): - __tablename__ = 'transaction_breakdown' - transaction_breakdown_id = Column(Integer, primary_key=True, autoincrement=True) - transaction_id = Column(Integer, ForeignKey('transaction.transaction_id')) - transaction_account_id = Column(Integer, ForeignKey('transaction_account.transaction_account_id')) - spent_amount = Column(Numeric(10, 2), nullable=False, default=0) - earned_amount = Column(Numeric(10, 2), nullable=False, default=0) - transaction = relationship("Transaction", back_populates="breakdowns") - -class TagAssignedToTransaction(Base): - __tablename__ = 'tag_assigned_to_transaction' - tag_assigned_to_transaction_id = Column(Integer, primary_key=True, autoincrement=True) - transaction_id = Column(Integer, ForeignKey('transaction.transaction_id')) - tag_id = Column(Integer, ForeignKey('tag.tag_id')) Index: backend_python/auth.py =================================================================== --- backend_python/auth.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) +++ backend_python/auth.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) @@ -0,0 +1,46 @@ +import jwt +from datetime import datetime, timedelta +from fastapi import HTTPException, status +from passlib.hash import bcrypt + +def hash_password(password: str) -> str: + return bcrypt.hash(password) + +def verify_password(password: str, hashed_password: str) -> bool: + return bcrypt.verify(password, hashed_password) + +def is_admin(email: str) -> bool: + return email in ["admin@fein.com"] + +def create_access_token(data: dict): + to_encode = data.copy() + expire = datetime.utcnow() + timedelta(minutes= 30) + to_encode.update({"exp": expire}) + + # Ensure `sub` is a string + if "sub" in to_encode and not isinstance(to_encode["sub"], str): + to_encode["sub"] = str(to_encode["sub"]) + + # Add other claims (e.g., email) + if "email" in to_encode and not isinstance(to_encode["email"], str): + raise ValueError("Email must be a string") + + encoded_jwt = jwt.encode(to_encode, "A1B2C3D4E5F6G7H8I9J0K", algorithm="HS256") + return encoded_jwt + +def decode_access_token(token: str): + try: + payload = jwt.decode(token, "A1B2C3D4E5F6G7H8I9J0K", algorithms=["HS256"]) + return payload + except jwt.ExpiredSignatureError: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Token has expired", + headers={"WWW-Authenticate": "Bearer"}, + ) + except jwt.PyJWTError: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid token", + headers={"WWW-Authenticate": "Bearer"}, + ) Index: backend_python/database.py =================================================================== --- backend_python/database.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) +++ backend_python/database.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) @@ -0,0 +1,36 @@ +import os +from dotenv import load_dotenv +from pathlib import Path +from sqlalchemy import create_engine +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker + +# Load environment variables +env_path = Path(__file__).resolve().parent.parent / '.env' +load_dotenv(dotenv_path=env_path) + +# Database URL +DATABASE_URL = os.getenv("DATABASE_URL") + +# Create the database engine +engine = create_engine(DATABASE_URL, pool_pre_ping=True) + +# Create a configured "Session" class +SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) + +# Base class for models +Base = declarative_base() + +try: + with engine.connect() as connection: + print("Database connection successful") +except Exception as e: + print(f"Failed to connect to the database: {e}") + +# Dependency to get the database session +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() Index: backend_python/main.py =================================================================== --- backend_python/main.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) +++ backend_python/main.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) @@ -0,0 +1,1095 @@ +from fastapi import FastAPI, Depends, HTTPException +from sqlalchemy.orm import Session +from .database import get_db +from .models import User, TransactionAccount, Transaction, TransactionBreakdown, Tag, TagAssignedToTransaction +from pydantic import BaseModel +from typing import List, Optional +from datetime import datetime +from fastapi.security import OAuth2PasswordBearer +from .auth import create_access_token, decode_access_token, is_admin, hash_password, verify_password +from sqlalchemy import func, literal_column, select + +# Initialize FastAPI app +app = FastAPI() + +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login/") + +# Pydantic schemas for validation and response models +class UserCreate(BaseModel): + user_name: str + email: str + password: str + +class UserResponse(BaseModel): + user_id: int + user_name: str + email: str + + class Config: + from_attributes = True + +class TransactionAccountCreate(BaseModel): + account_name: str + balance: float + +class TransactionAccountResponse(BaseModel): + transaction_account_id: int + account_name: str + balance: float + + class Config: + from_attributes = True + +class TransactionBreakdownResponse(BaseModel): + transaction_account_id: int + earned_amount: float + spent_amount: float + + class Config: + from_attributes = True + +class TransactionCreateRequest(BaseModel): + transaction_name: str + amount: float = 0.0 # Default to 0 if not provided + date: Optional[datetime] = None # Default to None, will use current time if not provided + tag_id: Optional[int] = None # Optional tag + target_account_id: int # Mandatory target account + breakdowns: Optional[List[TransactionBreakdownResponse]] = None # Optional list of breakdowns + + +class TransactionUpdate(BaseModel): + transaction_name: str = None + amount: float = None + net_amount: float = None + date: datetime = None + +class TransactionResponse(BaseModel): + transaction_id: int + transaction_name: str + amount: float + net_amount: float + date: datetime + + class Config: + json_encoders = { + datetime: lambda v: v.isoformat() # Serialize datetime as ISO 8601 string + } + from_attributes = True + +class TagCreate(BaseModel): + tag_name: str + +class TagResponse(BaseModel): + tag_id: int + tag_name: str + + class Config: + from_attributes = True + +class TagAssign(BaseModel): + transaction_id: int + tag_id: int + +class AuthRequest(BaseModel): + user_name: Optional[str] = None + email: str + password: str + +class AuthResponse(BaseModel): + access_token: str + token_type: str + +# Dependency to get the current user +def get_current_user( + token: str = Depends(oauth2_scheme), + db: Session = Depends(get_db) +): + """ + Retrieves the current user based on the access token. + """ + try: + payload = decode_access_token(token) + user_id = payload.get("sub") + if not user_id: + raise HTTPException( + status_code=401, + detail="Invalid authentication credentials", + headers={"WWW-Authenticate": "Bearer"} + ) + + user = db.query(User).filter(User.user_id == user_id).first() + if not user: + raise HTTPException( + status_code=401, + detail="Invalid authentication credentials", + headers={"WWW-Authenticate": "Bearer"} + ) + + return user + except Exception as e: + print(f"Error decoding token or fetching user: {e}") + raise HTTPException( + status_code=401, + detail="Invalid authentication credentials", + headers={"WWW-Authenticate": "Bearer"} + ) + +# Routes +@app.get("/") +def read_root(): + return {"message": "Welcome to the Fein Prototype API"} + +@app.post("/auth/register/", response_model=AuthResponse) +def register( + auth_request: AuthRequest, + db: Session = Depends(get_db) +): + # Check if email already exists + existing_user = db.query(User).filter(User.email == auth_request.email).first() + if existing_user: + raise HTTPException( + status_code=400, + detail="Email already registered" + ) + + # Hash password and create new user + hashed_password = hash_password(auth_request.password) + new_user = User( + user_name=auth_request.user_name, + email=auth_request.email, + password=hashed_password + ) + db.add(new_user) + db.commit() + db.refresh(new_user) + + # Return access token + access_token = create_access_token({"sub": new_user.user_id, "email": new_user.email}) + return {"access_token": access_token, "token_type": "bearer"} + +@app.post("/auth/login/", response_model=AuthResponse) +def login( + auth_request: AuthRequest, + db: Session = Depends(get_db) +): + # Verify email and password + user = db.query(User).filter(User.email == auth_request.email).first() + if not user or not verify_password(auth_request.password, user.password): + raise HTTPException( + status_code=401, + detail="Invalid email or password" + ) + + # Return access token + access_token = create_access_token({"sub": user.user_id, "email": user.email}) + return {"access_token": access_token, "token_type": "bearer"} + +@app.get("/admin/accounts/", response_model=List[TransactionAccountResponse]) +def admin_get_all_accounts( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Admin can fetch all transaction accounts. + """ + if not is_admin(user.email): + raise HTTPException( + status_code=403, + detail="Access denied" + ) + return db.query(TransactionAccount).all() + + +@app.post("/accounts/", response_model=TransactionAccountResponse) +def create_account( + account: TransactionAccountCreate, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + new_account = TransactionAccount( + account_name=account.account_name, + balance=account.balance, + user_id=user.user_id + ) + db.add(new_account) + db.commit() + db.refresh(new_account) + return new_account + +@app.get("/accounts/", response_model=List[TransactionAccountResponse]) +def get_accounts( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Admin can fetch all accounts, regular users only their accounts. + """ + query = db.query(TransactionAccount) + + if is_admin(user.email): + return query.all() + + return query.filter(TransactionAccount.user_id == user.user_id).all() + + +@app.post("/transactions/", response_model=TransactionResponse) +def create_transaction( + transaction_request: TransactionCreateRequest, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Admins can create transactions for any account; regular users only for their accounts. + Create a transaction and associate it with the user's accounts via breakdowns. + """ + # Admin bypasses ownership checks + if not is_admin(user.email): + # Validate target account ownership + target_account = db.query(TransactionAccount).filter( + TransactionAccount.transaction_account_id == transaction_request.target_account_id, + TransactionAccount.user_id == user.user_id + ).first() + if not target_account: + raise HTTPException( + status_code=403, + detail="Access denied to target account." + ) + + # Create transaction + new_transaction = Transaction( + transaction_name=transaction_request.transaction_name, + amount=transaction_request.amount, + net_amount=0.0, # Will be updated based on breakdowns + date=transaction_request.date or datetime.utcnow(), # Use current UTC time if not provided + ) + db.add(new_transaction) + db.commit() + db.refresh(new_transaction) + + # Associate a tag, if provided + if transaction_request.tag_id: + tag = db.query(Tag).filter(Tag.tag_id == transaction_request.tag_id).first() + if tag: + tag_assignment = TagAssignedToTransaction( + transaction_id=new_transaction.transaction_id, + tag_id=tag.tag_id + ) + db.add(tag_assignment) + + # Add breakdowns + net_amount = 0.0 + if transaction_request.breakdowns: + for breakdown in transaction_request.breakdowns: + # Validate breakdown account ownership + breakdown_account = db.query(TransactionAccount).filter( + TransactionAccount.transaction_account_id == breakdown.transaction_account_id, + TransactionAccount.user_id == user.user_id + ).first() + if not breakdown_account: + raise HTTPException( + status_code=403, + detail=f"Access denied to breakdown account {breakdown.transaction_account_id}." + ) + + # Create breakdown + new_breakdown = TransactionBreakdown( + transaction_id=new_transaction.transaction_id, + transaction_account_id=breakdown.transaction_account_id, + earned_amount=breakdown.earned_amount, + spent_amount=breakdown.spent_amount + ) + db.add(new_breakdown) + + # Calculate net amount + net_amount += breakdown.earned_amount - breakdown.spent_amount + + # Update transaction's net amount + new_transaction.net_amount = net_amount + db.commit() + db.refresh(new_transaction) + + return new_transaction + +@app.get("/transactions/", response_model=List[TransactionResponse]) +def get_transactions( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Fetch transactions based on user role, excluding placeholder transactions. + - Admin: Fetch all non-placeholder transactions. + - Regular User: Fetch non-placeholder transactions tied to the user's accounts via transaction breakdowns. + """ + query = db.query(Transaction) + + if is_admin(user.email): + # Admins see all non-placeholder transactions + transactions = ( + query + .filter(~Transaction.transaction_name.like("Tag_%_placeholder")) + .all() + ) + else: + transactions = ( + query + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(TransactionAccount.user_id == user.user_id) + .filter(~Transaction.transaction_name.like("Tag_%_placeholder")) # Exclude placeholders + .all() + ) + + return [ + { + **transaction.__dict__, + "date": transaction.date.isoformat() # Convert datetime to ISO 8601 string + } + for transaction in transactions + ] + +@app.get("/transactions/{transaction_id}", response_model=TransactionResponse) +def get_transaction_by_id( + transaction_id: int, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Retrieve a single transaction by its ID, ensuring access is restricted + to the transaction creator or an admin user. + """ + # If the user is an admin, they can access any transaction + if is_admin(user.email): + transaction = db.query(Transaction).filter(Transaction.transaction_id == transaction_id).first() + if not transaction: + raise HTTPException( + status_code=404, + detail="Transaction not found." + ) + return transaction + + # Otherwise, restrict access to the transaction creator + transaction = ( + db.query(Transaction) + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(Transaction.transaction_id == transaction_id) + .filter(TransactionAccount.user_id == user.user_id) + .first() + ) + if not transaction: + raise HTTPException( + status_code=404, + detail="Transaction not found or access denied." + ) + + return transaction + +@app.get("/transactions/{transaction_id}/breakdowns", response_model=List[TransactionBreakdownResponse]) +def get_transaction_breakdowns( + transaction_id: int, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Fetch transaction breakdowns for a specific transaction. + """ + breakdowns = ( + db.query(TransactionBreakdown) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(TransactionBreakdown.transaction_id == transaction_id) + .filter(TransactionAccount.user_id == user.user_id) + .all() + ) + + if not breakdowns: + raise HTTPException( + status_code=404, + detail="No breakdowns found for this transaction." + ) + + return breakdowns + + +@app.put("/transactions/{transaction_id}", response_model=TransactionResponse) +def update_transaction( + transaction_id: int, + transaction_update: TransactionUpdate, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Admins can update any transaction + Regular users update a transaction only if it belongs to the logged-in user. + """ + query = db.query(Transaction) + + if is_admin(user.email): + transaction = ( + query + .filter(Transaction.transaction_id == transaction_id) + .first() + ) + if not transaction: + raise HTTPException( + status_code=404, + detail="Transaction not found." + ) + else: + transaction = ( + query + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(Transaction.transaction_id == transaction_id) + .filter(TransactionAccount.user_id == user.user_id) + .first() + ) + if not transaction: + raise HTTPException( + status_code=404, + detail="Transaction not found or access denied." + ) + + # Update transaction fields + for key, value in transaction_update.dict(exclude_unset=True).items(): + setattr(transaction, key, value) + + db.commit() + db.refresh(transaction) + return transaction + +@app.delete("/transactions/{transaction_id}") +def delete_transaction( + transaction_id: int, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Admins can delete any transaction + Regular users can delete a transaction only if it belongs to the logged-in user. + """ + query = db.query(Transaction) + + if is_admin(user.email): + transaction = ( + query + .filter(Transaction.transaction_id == transaction_id) + .first() + ) + if not transaction: + raise HTTPException( + status_code=404, + detail="Transaction not found." + ) + else: + transaction = ( + query + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(Transaction.transaction_id == transaction_id) + .filter(TransactionAccount.user_id == user.user_id) + .first() + ) + if not transaction: + raise HTTPException( + status_code=404, + detail="Transaction not found or access denied." + ) + + db.delete(transaction) + db.commit() + return {"message": "Transaction deleted successfully"} + + +@app.post("/tags/", response_model=TagResponse) +def create_tag( + tag: TagCreate, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Create a tag associated with the logged-in user by linking it to a placeholder transaction. + """ + # Create the tag + new_tag = Tag(tag_name=tag.tag_name) + db.add(new_tag) + db.commit() + db.refresh(new_tag) + + # Create a dummy transaction linked to the user's first account + user_account = ( + db.query(TransactionAccount) + .filter(TransactionAccount.user_id == user.user_id) + .first() + ) + if not user_account: + raise HTTPException( + status_code=403, + detail="No account available to associate with the tag." + ) + + # Associate the tag with a dummy transaction for the user + dummy_transaction = Transaction( + transaction_name=f"Tag_{new_tag.tag_id}_placeholder", + amount=0, + net_amount=0, + date=datetime.utcnow(), + ) + db.add(dummy_transaction) + db.commit() + db.refresh(dummy_transaction) + + # Link the dummy transaction to the user's account + dummy_breakdown = TransactionBreakdown( + transaction_id=dummy_transaction.transaction_id, + transaction_account_id=user_account.transaction_account_id, + earned_amount=0, + spent_amount=0, + ) + db.add(dummy_breakdown) + + # Associate the tag with the dummy transaction + tag_assignment = TagAssignedToTransaction( + transaction_id=dummy_transaction.transaction_id, + tag_id=new_tag.tag_id, + ) + db.add(tag_assignment) + db.commit() + + return new_tag + +@app.get("/tags/", response_model=List[TagResponse]) +def get_tags( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Admins can fetch all tags + Regular users can retrieve tags accessible to the logged-in user based on their transactions. + """ + if is_admin(user.email): + return db.query(Tag).all() + + accessible_tags = ( + db.query(Tag) + .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id) + .join(Transaction, TagAssignedToTransaction.transaction_id == Transaction.transaction_id) + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(TransactionAccount.user_id == user.user_id) + .distinct() + .all() + ) + return accessible_tags + +@app.post("/tags/assign/", response_model=dict) +def assign_tag_to_transaction( + tag_assign: TagAssign, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Assign a tag to a transaction. + - Admins can assign any tag to any transaction. + - Regular users can assign a tag if: + - The transaction belongs to them. + - The tag is accessible (created by them or linked to their transactions). + """ + # Ensure the transaction belongs to the logged-in user + transaction = ( + db.query(Transaction) + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(Transaction.transaction_id == tag_assign.transaction_id) + .filter(TransactionAccount.user_id == user.user_id) + .first() + ) + if not transaction: + raise HTTPException(status_code=404, detail="Transaction not found or access denied.") + + # Ensure the tag is accessible to the logged-in user + tag_accessible = ( + db.query(Tag) + .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id, isouter=True) + .join(Transaction, TagAssignedToTransaction.transaction_id == Transaction.transaction_id, isouter=True) + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id, isouter=True) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id, isouter=True) + .filter(Tag.tag_id == tag_assign.tag_id) + .filter( + (TransactionAccount.user_id == user.user_id) | # Tag linked to the user's transactions + (TransactionAccount.user_id.is_(None)) # Newly created tag not yet assigned + ) + .first() + ) + if not tag_accessible: + raise HTTPException( + status_code=404, + detail="Access denied to the tag." + ) + + # Check if the tag is already assigned to the transaction + existing_assignment = ( + db.query(TagAssignedToTransaction) + .filter( + TagAssignedToTransaction.transaction_id == tag_assign.transaction_id, + TagAssignedToTransaction.tag_id == tag_assign.tag_id, + ) + .first() + ) + if existing_assignment: + raise HTTPException( + status_code=400, + detail="Tag already assigned to this transaction." + ) + + # Assign the tag to the transaction + assignment = TagAssignedToTransaction( + transaction_id=tag_assign.transaction_id, + tag_id=tag_assign.tag_id, + ) + db.add(assignment) + db.commit() + + return {"message": "Tag assigned to transaction successfully"} + +@app.get("/tags/transaction/{transaction_id}", response_model=List[TagResponse]) +def get_transaction_tags_for_user( + transaction_id: int, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Retrieve tags for a specific transaction. + - Admins can retrieve tags for any transaction. + - Regular users can retrieve tags if the transaction belongs to them. + """ + # Admins can access tags for any transaction + if is_admin(user.email): + tags = ( + db.query(Tag) + .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id) + .filter(TagAssignedToTransaction.transaction_id == transaction_id) + .all() + ) + return tags + + # Check if the transaction belongs to the user + transaction = ( + db.query(Transaction) + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(Transaction.transaction_id == transaction_id) + .filter(TransactionAccount.user_id == user.user_id) + .first() + ) + if not transaction: + raise HTTPException( + status_code=403, + detail="Access denied" + ) + + # Retrieve tags for the transaction + tags = ( + db.query(Tag) + .join(TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id) + .filter(TagAssignedToTransaction.transaction_id == transaction_id) + .all() + ) + + return tags + +@app.get("/reports/total-spending", response_model=dict) +def get_total_spending( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Calculate and return total spending for the logged-in user. + - Admins can view total spending for all users. + """ + try: + query = db.query( + func + .sum(Transaction.amount) + .label("total_spent") + ) + + if is_admin(user.email): + # Admin: Total spending for all users + total_spent = ( + query + .filter(Transaction.amount > 0) + .scalar() + ) + else: + # Regular User: Total spending for their accounts + total_spent = ( + query + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter(TransactionAccount.user_id == user.user_id) + .filter(Transaction.amount > 0) + .scalar() + ) + + return {"total_spent": total_spent or 0.0} + except Exception as e: + print(f"Error calculating total spending: {e}") + raise HTTPException( + status_code=500, + detail="Failed to calculate total spending." + ) + +@app.get("/reports/spending-by-category", response_model=dict) +def get_spending_by_category( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Calculate and return spending grouped by category (tags) for the logged-in user. + - Admins can view spending by category for all users. + """ + try: + # Base query + query = db.query( + Tag.tag_name, + func.sum(Transaction.amount).label("total_spent") + ).join( + TagAssignedToTransaction, Tag.tag_id == TagAssignedToTransaction.tag_id + ).join( + Transaction, TagAssignedToTransaction.transaction_id == Transaction.transaction_id + ).filter( + Transaction.amount > 0 # Include only positive amounts + ) + + # Apply filters for regular users + if not is_admin(user.email): + query = query.join( + TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id + ).join( + TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id + ).filter( + TransactionAccount.user_id == user.user_id + ) + + # Group by tag and calculate the total spending for each category + spending_by_category = query.group_by(Tag.tag_name).all() + + # Prepare the response as a dictionary + response = {row.tag_name: float(row.total_spent or 0) for row in spending_by_category} + return {"spending_by_category": response} + + except Exception as e: + print(f"Error calculating spending by category: {e}") + raise HTTPException( + status_code=500, + detail="Failed to calculate spending by category." + ) + +@app.get("/reports/spending-by-date-range", response_model=dict) +def get_spending_by_date_range( + start_date: str, # Expecting date in 'YYYY-MM-DD' format + end_date: str, + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Calculate and return spending within a specified date range for the logged-in user. + - Admins can view spending within the date range for all users. + """ + try: + # Convert input dates to `datetime` + start_date_parsed = datetime.strptime(start_date, "%Y-%m-%d") + end_date_parsed = datetime.strptime(end_date, "%Y-%m-%d") + + # Query base + query = db.query(func.sum(Transaction.amount).label("total_spent")) + + if is_admin(user.email): + # Admin: Total spending for all users within the date range + total_spent = ( + query + .filter( + Transaction.date >= start_date_parsed, + Transaction.date <= end_date_parsed, + Transaction.amount > 0 + ) + .scalar() + ) + else: + # Regular User: Total spending within the date range for their accounts + total_spent = ( + query + .join(TransactionBreakdown, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(TransactionAccount, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .filter( + TransactionAccount.user_id == user.user_id, + Transaction.date >= start_date_parsed, + Transaction.date <= end_date_parsed, + Transaction.amount > 0 + ) + .scalar() + ) + + # Return result + return {"total_spent": total_spent or 0.0} + except Exception as e: + print(f"Error calculating spending by date range: {e}") + raise HTTPException( + status_code=500, + detail="Failed to calculate spending by date range." + ) + +@app.get("/reports/exceeding-transactions", response_model=List[dict]) +def get_exceeding_transactions( + account_name: Optional[str] = None, # Allow filtering by account name + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Retrieve a list of transactions that exceeded the balance of an account, sorted chronologically. + - Admins can view for all users. + - Regular users can view for their own accounts. + """ + # Define the subquery to calculate `calculated_balance` using a window function + subquery = ( + db.query( + Transaction.transaction_id, + Transaction.transaction_name, + Transaction.date.label("transaction_date"), + TransactionAccount.account_name, + User.user_id, + User.user_name, + TransactionBreakdown.spent_amount.label("transaction_amount"), + func.sum(TransactionBreakdown.earned_amount - TransactionBreakdown.spent_amount) + .over( + partition_by=TransactionBreakdown.transaction_account_id, + order_by=Transaction.date + ) + .label("calculated_balance"), + ) + .join(TransactionAccount, TransactionAccount.transaction_account_id == TransactionBreakdown.transaction_account_id) + .join(User, TransactionAccount.user_id == User.user_id) + .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .subquery() + ) + + query = db.query( + subquery.c.transaction_id, + subquery.c.transaction_name, + subquery.c.transaction_date, + subquery.c.account_name, + subquery.c.user_id, + subquery.c.user_name, + subquery.c.transaction_amount, + subquery.c.calculated_balance, + ).filter( + subquery.c.transaction_amount > subquery.c.calculated_balance, # Filter where transaction amount exceeds balance + subquery.c.transaction_amount > 0, # Filter for positive transactions + ) + + if account_name: + query = query.filter(subquery.c.account_name == account_name) + + # Apply user-specific filtering for non-admins + if not is_admin(user.email): + query = query.filter(subquery.c.user_id == user.user_id) + + # Order results + query = query.order_by( + subquery.c.user_id, + subquery.c.account_name, + subquery.c.transaction_date.desc(), + ) + + # Execute the query and fetch results + results = query.all() + + if not results: + return [] + + # Prepare response + response = [ + { + "user_id": row.user_id, + "user_name": row.user_name, + "account_name": row.account_name, + "transaction_id": row.transaction_id, + "transaction_name": row.transaction_name, + "transaction_amount": row.transaction_amount, + "transaction_date": row.transaction_date, + "calculated_balance": row.calculated_balance, + } + for row in results + ] + + return response + +@app.get("/reports/exceeding-current-balance", response_model=List[dict]) +def get_exceeding_current_balance( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Retrieve a list of transactions that exceed the current balance of accounts. + - Admins can view for all users. + - Regular users can view for their own accounts. + """ + query = ( + db.query( + User.user_id, + User.user_name, + TransactionAccount.account_name, + TransactionAccount.balance.label("current_balance"), + Transaction.transaction_id, + Transaction.transaction_name, + TransactionBreakdown.spent_amount.label("transaction_amount"), + Transaction.date.label("transaction_date"), + ) + .join(TransactionAccount, TransactionAccount.user_id == User.user_id) + .join(TransactionBreakdown, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .filter(TransactionBreakdown.spent_amount > TransactionAccount.balance) # Transactions exceeding account balance + .filter(TransactionBreakdown.spent_amount > 0) # Positive transactions only + ) + + # Apply user-specific filtering for non-admins + if not is_admin(user.email): + query = query.filter(TransactionAccount.user_id == user.user_id) + + # Order results + results = query.order_by(User.user_id, TransactionAccount.account_name, Transaction.date.desc()).all() + + # Prepare response + return [ + { + "user_id": row.user_id, + "user_name": row.user_name, + "account_name": row.account_name, + "current_balance": float(row.current_balance), + "transaction_id": row.transaction_id, + "transaction_name": row.transaction_name, + "transaction_amount": float(row.transaction_amount), + "transaction_date": row.transaction_date, + } + for row in results + ] + +@app.get("/reports/exceeding-total-balances", response_model=List[dict]) +def get_exceeding_total_balances( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Retrieve a chronological list of transactions that exceed the calculated total balances for all accounts. + - Admins can view for all users. + - Regular users can view for their own accounts. + """ + # Subquery to calculate the running balance using a window function + subquery = ( + db.query( + User.user_id.label("user_id"), + User.user_name.label("user_name"), + Transaction.transaction_id.label("transaction_id"), + Transaction.transaction_name.label("transaction_name"), + Transaction.date.label("transaction_date"), + TransactionBreakdown.spent_amount.label("transaction_amount"), + func.sum(TransactionBreakdown.earned_amount - TransactionBreakdown.spent_amount) + .over(partition_by=User.user_id, order_by=Transaction.date) + .label("calculated_total_balance") + ) + .join(TransactionAccount, TransactionAccount.user_id == User.user_id) + .join(TransactionBreakdown, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .filter(TransactionBreakdown.spent_amount > 0) # Only positive transactions + .subquery() + ) + + # Outer query to filter transactions exceeding the calculated balance + query = db.query( + subquery.c.user_id, + subquery.c.user_name, + subquery.c.transaction_id, + subquery.c.transaction_name, + subquery.c.transaction_date, + subquery.c.transaction_amount, + subquery.c.calculated_total_balance + ).filter( + subquery.c.transaction_amount > subquery.c.calculated_total_balance # Exceeds the total balance + ) + + # Apply user-specific filtering for non-admin users + if not is_admin(user.email): + query = query.filter(subquery.c.user_id == user.user_id) + + # Execute the query and return results + results = query.order_by(subquery.c.user_id, subquery.c.transaction_date.desc()).all() + + return [ + { + "user_id": row.user_id, + "user_name": row.user_name, + "transaction_id": row.transaction_id, + "transaction_name": row.transaction_name, + "transaction_date": row.transaction_date, + "transaction_amount": float(row.transaction_amount), + "calculated_total_balance": float(row.calculated_total_balance), + } + for row in results + ] + +@app.get("/reports/exceeding-user-total-balance", response_model=List[dict]) +def get_exceeding_user_total_balance( + user: User = Depends(get_current_user), + db: Session = Depends(get_db) +): + """ + Retrieve a list of users whose transactions exceed the total balance of all their accounts. + - Admins can view results for all users. + - Regular users can only see their own data. + """ + # Subquery to calculate the total balance for each user + total_balance_subquery = ( + db.query( + TransactionAccount.user_id.label("user_id"), + func.sum(TransactionAccount.balance).label("total_balance") + ) + .group_by(TransactionAccount.user_id) + .subquery() + ) + + # Main query + query = ( + db.query( + User.user_id, + User.user_name, + func.sum(TransactionBreakdown.spent_amount).label("total_transaction_amount"), + total_balance_subquery.c.total_balance.label("user_total_balance") + ) + .join(TransactionAccount, TransactionAccount.user_id == User.user_id) + .join(TransactionBreakdown, TransactionBreakdown.transaction_account_id == TransactionAccount.transaction_account_id) + .join(Transaction, Transaction.transaction_id == TransactionBreakdown.transaction_id) + .join(total_balance_subquery, total_balance_subquery.c.user_id == User.user_id) + .filter(Transaction.date <= func.current_date()) # Only transactions up to the current date + .group_by(User.user_id, User.user_name, total_balance_subquery.c.total_balance) + .having(func.sum(TransactionBreakdown.spent_amount) > total_balance_subquery.c.total_balance) # Exceeds total balance + .order_by(User.user_id) + ) + + # Apply user-specific filtering for non-admins + if not is_admin(user.email): + query = query.filter(User.user_id == user.user_id) + + # Execute the query + results = query.all() + + # Prepare response + return [ + { + "user_id": row.user_id, + "user_name": row.user_name, + "total_transaction_amount": float(row.total_transaction_amount), + "user_total_balance": float(row.user_total_balance), + } + for row in results + ] + Index: backend_python/models.py =================================================================== --- backend_python/models.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) +++ backend_python/models.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) @@ -0,0 +1,57 @@ +from sqlalchemy import Column, Integer, String, Numeric, ForeignKey, DateTime, Boolean +from sqlalchemy.orm import relationship +from sqlalchemy.ext.declarative import declarative_base + +Base = declarative_base() + +# Define Models +class User(Base): + __tablename__ = 'user' + user_id = Column(Integer, primary_key=True, autoincrement=True) + user_name = Column(String(50), nullable=False) + email = Column(String(100), unique=True, nullable=False) + password = Column(String(255), nullable=False) + accounts = relationship("TransactionAccount", back_populates="user") + +class TransactionAccount(Base): + __tablename__ = 'transaction_account' + transaction_account_id = Column(Integer, primary_key=True, autoincrement=True) + account_name = Column(String(50), nullable=False) + balance = Column(Numeric(10, 2), default=0, nullable=False) + user_id = Column(Integer, ForeignKey('user.user_id')) + user = relationship("User", back_populates="accounts") + +class Tag(Base): + __tablename__ = 'tag' + tag_id = Column(Integer, primary_key=True, autoincrement=True) + tag_name = Column(String(50), nullable=False) + transactions = relationship( + "Transaction", secondary="tag_assigned_to_transaction", back_populates="tags" + ) + +class Transaction(Base): + __tablename__ = 'transaction' + transaction_id = Column(Integer, primary_key=True, autoincrement=True) + transaction_name = Column(String(100), nullable=False) + amount = Column(Numeric(10, 2), nullable=False) + net_amount = Column(Numeric(10, 2), nullable=False) + date = Column(DateTime(timezone=True), nullable=False) + breakdowns = relationship("TransactionBreakdown", back_populates="transaction") + tags = relationship( + "Tag", secondary="tag_assigned_to_transaction", back_populates="transactions" + ) + +class TransactionBreakdown(Base): + __tablename__ = 'transaction_breakdown' + transaction_breakdown_id = Column(Integer, primary_key=True, autoincrement=True) + transaction_id = Column(Integer, ForeignKey('transaction.transaction_id')) + transaction_account_id = Column(Integer, ForeignKey('transaction_account.transaction_account_id')) + spent_amount = Column(Numeric(10, 2), nullable=False, default=0) + earned_amount = Column(Numeric(10, 2), nullable=False, default=0) + transaction = relationship("Transaction", back_populates="breakdowns") + +class TagAssignedToTransaction(Base): + __tablename__ = 'tag_assigned_to_transaction' + tag_assigned_to_transaction_id = Column(Integer, primary_key=True, autoincrement=True) + transaction_id = Column(Integer, ForeignKey('transaction.transaction_id')) + tag_id = Column(Integer, ForeignKey('tag.tag_id')) Index: instructions.txt =================================================================== --- instructions.txt (revision 0ba4abde554eb4472ff4126fe101af6200437345) +++ instructions.txt (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) @@ -1,5 +1,6 @@ +Note: Database URL needs to start with `postgresql://` instead of `postgres://` 1. Start the backend: -uvicorn backend-python.main:app --host 0.0.0.0 --port 8000 --reload +uvicorn backend_python.main:app --host 0.0.0.0 --port 8000 --reload 2. Start the prototype application in the CommandLine: -python -m prototype-cli-app.cli_app +python -m prototype_cli_app.cli_app Index: ototype-cli-app/cli_app.py =================================================================== --- prototype-cli-app/cli_app.py (revision 0ba4abde554eb4472ff4126fe101af6200437345) +++ (revision ) @@ -1,640 +1,0 @@ -import requests -import jwt -from app.auth import is_admin, decode_access_token - -BASE_URL = "http://localhost:8000" -access_token = None # Global variable for the access token - -def main_menu(): - """Displays the main menu for registration and login.""" - global access_token - while True: - print("\nMain Menu") - print("1. Register") - print("2. Login") - print("3. Exit") - choice = input("Choose an option: ") - - if choice == "1": - register() - elif choice == "2": - token = login() - if token: - access_token = token - handle_menu_after_login() - elif choice == "3": - print("Exiting...") - break - else: - print("Invalid choice. Please try again.") - -def register(): - """Handles user registration.""" - print("\nRegister") - user_name = input("Enter your username: ") - email = input("Enter your email: ") - password = input("Enter your password: ") - - response = requests.post(f"{BASE_URL}/auth/register/", json={ - "user_name": user_name, - "email": email, - "password": password - }) - - if response.status_code == 200: - print("Registration successful.") - else: - print("Registration failed. Please try again.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def login(): - """Handles user login and returns the access token.""" - print("\nLog in") - email = input("Enter your email: ") - password = input("Enter your password: ") - - response = requests.post(f"{BASE_URL}/auth/login/", json={ - "email": email, - "password": password - }) - - if response.status_code == 200: - data = response.json() - print(f"Log in successful. Access Token: {data['access_token']}") - return data["access_token"] - else: - print("Invalid login credentials.") - return None - -def handle_menu_after_login(): - """Routes the user to the appropriate menu based on their role.""" - global access_token - # Decode the JWT to extract user information - try: - payload = decode_access_token(access_token) - user_id = payload.get("sub") # Extract user_id from the token - email = payload.get("email") # Extract email from the token - - if not user_id or not email: - raise ValueError("Token is missing required fields.") - - print(f"User ID from token: {user_id}, Email: {email}") - - # Check if the user is an admin - if is_admin(email): - admin_menu() - else: - user_menu() - except jwt.ExpiredSignatureError: - print("Session expired. Please log in again.") - access_token = None - except ValueError as e: - print(f"Token validation error: {str(e)}. Logging out.") - access_token = None - except Exception as e: - print(f"An error occurred: {str(e)}. Logging out.") - access_token = None - -def admin_menu(): - """Displays the admin menu.""" - global access_token - while True: - print("\nAdmin Menu") - print("1. View All Transaction Accounts") - print("2. Log Out") - choice = input("Choose an option: ") - - if choice == "1": - admin_view_all_accounts() - elif choice == "2": - print("Logging out...") - access_token = None - break - else: - print("Invalid choice. Please try again.") - -def admin_view_all_accounts(): - """Fetches and displays all transaction accounts for admin users.""" - headers = {"Authorization": f"Bearer {access_token}"} - print("\nAll Transaction Accounts") - response = requests.get(f"{BASE_URL}/admin/accounts/", headers=headers) - - if response.status_code == 200: - accounts = response.json() - for account in accounts: - print(f"Account ID: {account['transaction_account_id']}\n" - +f" Name: {account['account_name']}\n" - +f" Balance: {account['balance']}\n") - else: - print("Failed to retrieve accounts.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def user_menu(): - """Displays the user menu.""" - global access_token - while True: - print("\nUser Menu") - print("1. Add Transaction Account") - print("2. View Transaction Accounts") - print("3. Add Transaction") - print("4. View Transactions") - print("5. Modify Transaction") - print("6. Delete Transaction") - print("7. Add Tag") - print("8. View Tags") - print("9. Assign Tag to Transaction") - print("10. View Transaction Tags") - print("11. View Reports") - print("12. Log Out") - choice = input("Choose an option: ") - - if choice == "1": - add_transaction_account() - elif choice == "2": - view_transaction_accounts() - elif choice == "3": - add_transaction() - elif choice == "4": - view_transactions() - elif choice == "5": - modify_transaction() - elif choice == "6": - delete_transaction() - elif choice == "7": - add_tag() - elif choice == "8": - view_tags() - elif choice == "9": - assign_tag_to_transaction() - elif choice == "10": - view_transaction_tags() - elif choice == "11": - reports_menu() - elif choice == "12": - print("Logging out...") - access_token = None - break - else: - print("Invalid choice. Please try again.") - -def add_transaction_account(): - headers = {"Authorization": f"Bearer {access_token}"} - print("\nAdd Transaction Account") - account_name = input("Enter account name: ") - balance = float(input("Enter balance: ")) - - response = requests.post(f"{BASE_URL}/accounts/", json={ - "account_name": account_name, - "balance": balance - }, headers=headers) - - if response.status_code == 200: - print("Transaction account added successfully.") - else: - print("Failed to add transaction account.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def view_transaction_accounts(): - headers = {"Authorization": f"Bearer {access_token}"} - print("\nTransaction Accounts") - response = requests.get(f"{BASE_URL}/accounts/", headers=headers) - - if response.status_code == 200: - accounts = response.json() - for account in accounts: - print(f"Account ID: {account['transaction_account_id']}") - print(f"Account Name: {account['account_name']}") - print(f"Balance: {account['balance']}") - print() - else: - print("Failed to retrieve transaction accounts.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def add_transaction(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nAdd Transaction") - transaction_name = input("Enter transaction name: ") - amount = float(input("Enter amount (default 0): ") or 0) - date = input("Enter date (YYYY-MM-DDTHH:MM:SS+HH:MM: ") - target_account_id = int(input("Enter target transaction account ID: ")) - tag_id = int(input("Enter tag ID (leave blank for none): ") or 0) - - breakdowns = [] - add_breakdown = input("Do you want to add breakdowns? (yes/no): ").lower() - while add_breakdown == "yes": - breakdown_account_id = int(input("Enter breakdown account ID: ")) - earned_amount = float(input("Enter earned amount (default 0): ") or 0) - spent_amount = float(input("Enter spent amount (default 0): ") or 0) - breakdowns.append({ - "transaction_account_id": breakdown_account_id, - "earned_amount": earned_amount, - "spent_amount": spent_amount - }) - add_breakdown = input("Add another breakdown? (yes/no): ").lower() - - payload = { - "transaction_name": transaction_name, - "amount": amount, - "date": date, - "target_account_id": target_account_id, - "tag_id": tag_id or None, - "breakdowns": breakdowns - } - - response = requests.post(f"{BASE_URL}/transactions/", json=payload, headers=headers) - - if response.status_code == 200: - print("Transaction added successfully.") - else: - print("Failed to add transaction.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def view_transactions(): - headers = {"Authorization": f"Bearer {access_token}"} - print("\nYour Transactions:") - response = requests.get(f"{BASE_URL}/transactions/", headers=headers) - - if response.status_code == 200: - transactions = response.json() - for transaction in transactions: - print(f"Transaction ID: {transaction['transaction_id']}") - print(f" Name: {transaction['transaction_name']}") - print(f" Amount: {transaction['amount']}") - print(f" Net Amount: {transaction['net_amount']}") - print(f" Date: {transaction['date']}") - print(f" Breakdowns:") - - breakdown_response = requests.get( - f"{BASE_URL}/transactions/{transaction['transaction_id']}/breakdowns", - headers=headers - ) - - if breakdown_response.status_code == 200: - breakdowns = breakdown_response.json() - for breakdown in breakdowns: - print(f" Account ID: {breakdown['transaction_account_id']}") - print(f" Earned: {breakdown['earned_amount']}") - print(f" Spent: {breakdown['spent_amount']}") - else: - print(f" Could not fetch breakdowns. Error: {breakdown_response.json().get('detail', 'Unknown error')}") - print() - else: - print("Failed to retrieve transactions.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def modify_transaction(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nModify Transaction") - transaction_id = int(input("Enter transaction ID to modify: ")) - transaction_name = input("Enter new transaction name (or leave blank): ") - amount = input("Enter new amount (or leave blank): ") - net_amount = input("Enter new net amount (or leave blank): ") - date = input("Enter new date (YYYY-MM-DDTHH:MM:SS+HH:MM) (or leave blank): ") - - # Build the request payload - data = {} - if transaction_name: - data["transaction_name"] = transaction_name - if amount: - data["amount"] = float(amount) - if net_amount: - data["net_amount"] = float(net_amount) - if date: - data["date"] = date - - # Make the PUT request with the Authorization header - response = requests.put(f"{BASE_URL}/transactions/{transaction_id}", json=data, headers=headers) - - if response.status_code == 200: - print("Transaction modified successfully.") - else: - print("Failed to modify transaction.") - error_message = response.json().get('detail', 'Unknown error') - print(f"Error: {error_message}") - -def delete_transaction(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nDelete Transaction") - transaction_id = int(input("Enter transaction ID to delete: ")) - - response = requests.delete(f"{BASE_URL}/transactions/{transaction_id}", headers=headers) - - if response.status_code == 200: - print("Transaction deleted successfully.") - else: - print("Failed to delete transaction.") - error_message = response.json().get('detail', 'Unknown error') - print(f"Error: {error_message}") - -def add_tag(): - """Handles creating a new tag for the logged-in user.""" - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nAdd Tag") - tag_name = input("Enter tag name: ") - - response = requests.post(f"{BASE_URL}/tags/", json={"tag_name": tag_name}, headers=headers) - - if response.status_code == 200: - print("Tag added successfully.") - else: - print("Failed to add tag.") - error_message = response.json().get('detail', 'Unknown error') - print(f"Error: {error_message}") - -def view_tags(): - """Retrieve and display tags accessible to the logged-in user.""" - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nYour Tags") - response = requests.get(f"{BASE_URL}/tags/", headers=headers) - - if response.status_code == 200: - tags = response.json() - for tag in tags: - print(f"ID: {tag['tag_id']}, Name: {tag['tag_name']}") - else: - print(f"Failed to retrieve tags. Error: {response.json().get('detail', 'Unknown error')}") - -def assign_tag_to_transaction(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nAssign Tag to Transaction") - transaction_id = int(input("Enter transaction ID: ")) - tag_id = int(input("Enter tag ID: ")) - - response = requests.post(f"{BASE_URL}/tags/assign/", json={ - "transaction_id": transaction_id, - "tag_id": tag_id - }, headers=headers) - - if response.status_code == 200: - print("Tag assigned successfully.") - else: - print("Failed to assign tag.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - -def view_transaction_tags(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nView Transaction Tags") - transaction_id = int(input("Enter transaction ID to view its tags: ")) - - response = requests.get(f"{BASE_URL}/transactions/{transaction_id}", headers=headers) - - if response.status_code == 200: - transaction = response.json() - print(f"Transaction: {transaction['transaction_name']}") - - # Fetch tags linked to the transaction - tags_response = requests.get(f"{BASE_URL}/tags/transaction/{transaction_id}", headers=headers) - if tags_response.status_code == 200: - tags = tags_response.json() - for tag in tags: - print(f"Tag ID: {tag['tag_id']}, Name: {tag['tag_name']}") - else: - print(f"Failed to retrieve tags. Error: {tags_response.json().get('detail', 'Unknown error')}") - else: - print(f"Failed to retrieve transaction. Error: {response.json().get('detail', 'Unknown error')}") - -def reports_menu(): - """ - Displays the reports menu. - """ - global access_token - - while True: - print("\nReports Menu") - print("0. Go back to User Menu") - print("1. View Total Spending") - print("2. View Spending by Category") - print("3. View Spending by Date Range") - print("4. View Transactions Exceeding Account Balance") - print("5. View Transactions Exceeding Account Balance Currently") - print("6. View Chronological List of Transactions Exceeding Total Account Balances") - print("7. View Transactions Exceeding User's Total Current Balance") - choice = input("Choose a report option: ") - - if choice == "1": - get_total_spending() - elif choice == "2": - get_spending_by_category() - elif choice == "3": - get_spending_by_date_range() - elif choice == "4": - get_exceeding_account_balance() - elif choice == "5": - get_exceeding_current_balance() - elif choice == "6": - get_exceeding_total_balances() - elif choice == "7": - get_exceeding_user_total_balance() - elif choice == "0": - break - else: - print("Invalid choice. Please try again.") - -def get_total_spending(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nTotal Spending Report") - try: - response = requests.get(f"{BASE_URL}/reports/total-spending", headers=headers) - - if response.status_code == 200: - data = response.json() - total_spent = data.get("total_spent", 0.0) - print(f"Your total spending is: {float(total_spent):,.2f}") - else: - print("Failed to fetch the total spending report.") - error_message = response.json().get("detail", "Unknown error") - print(f"Error: {error_message}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - -def get_spending_by_category(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nSpending by Category Report") - try: - response = requests.get(f"{BASE_URL}/reports/spending-by-category", headers=headers) - - if response.status_code == 200: - data = response.json() - spending_by_category = data.get("spending_by_category", {}) - - if not spending_by_category: - print("No spending data found by category.") - else: - print("\nSpending by Category:") - print("-" * 50) - for category, amount in spending_by_category.items(): - print(f"Category: {category}") - print(f"Total Spending: {float(amount):,.2f}") - print("-" * 50) - else: - print("Failed to fetch the spending by category report.") - error_message = response.json().get("detail", "Unknown error") - print(f"Error: {error_message}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - -def get_spending_by_date_range(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nSpending by Date Range Report") - start_date = input("Enter start date (YYYY-MM-DD): ") - end_date = input("Enter end date (YYYY-MM-DD): ") - - try: - response = requests.get( - f"{BASE_URL}/reports/spending-by-date-range", - params={"start_date": start_date, "end_date": end_date}, - headers=headers - ) - - if response.status_code == 200: - data = response.json() - total_spent = data.get("total_spent", 0.0) - print("\nSpending by Date Range:") - print(f"From: {start_date} To: {end_date}") - print(f"Total Spending: {float(total_spent):,.2f}") - else: - print("Failed to fetch the spending by date range report.") - error_message = response.json().get("detail", "Unknown error") - print(f"Error: {error_message}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - -def get_exceeding_account_balance(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nTransactions Exceeding Account Balance Report") - account_name = input("Enter account name (leave blank for all accounts): ") - params = {"account_name": account_name} if account_name else {} - - try: - response = requests.get( - f"{BASE_URL}/reports/exceeding-transactions", - params=params, - headers=headers - ) - - if response.status_code == 200: - records = response.json() - if not records: - print("No transactions exceeding account balance were found.") - else: - print("\nExceeding Transactions:") - for record in records: - print("-" * 50) - print(f"User ID: {record['user_id']}") - print(f"User Name: {record['user_name']}") - print(f"Account Name: {record['account_name']}") - print(f"Transaction ID: {record['transaction_id']}") - print(f"Transaction Name: {record['transaction_name']}") - print(f"Transaction Amount: {float(record['transaction_amount']):,.2f}") - print(f"Transaction Date: {record['transaction_date']}") - print(f"Calculated Balance: {float(record['calculated_balance']):,.2f}") - print("-" * 50) - else: - print("Failed to fetch the exceeding transactions report.") - print(f"Error: {response.json().get('detail', 'Unknown error')}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - -def get_exceeding_current_balance(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nTransactions Exceeding Current Account Balance Report") - try: - response = requests.get(f"{BASE_URL}/reports/exceeding-current-balance", headers=headers) - - if response.status_code == 200: - records = response.json() - if not records: - print("No transactions exceeding current account balance were found.") - else: - print("\nTransactions Exceeding Current Balance:") - for record in records: - print("-" * 50) - print(f"User ID: {record['user_id']}") - print(f"User Name: {record['user_name']}") - print(f"Account Name: {record['account_name']}") - print(f"Current Balance: {record['current_balance']:,.2f}") - print(f"Transaction ID: {record['transaction_id']}") - print(f"Transaction Name: {record['transaction_name']}") - print(f"Transaction Amount: {record['transaction_amount']:,.2f}") - print(f"Transaction Date: {record['transaction_date']}") - print("-" * 50) - else: - print("Failed to fetch the report.") - error_message = response.json().get("detail", "Unknown error") - print(f"Error: {error_message}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - -def get_exceeding_total_balances(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nChronological List of Transactions Exceeding Total Account Balances") - try: - response = requests.get(f"{BASE_URL}/reports/exceeding-total-balances", headers=headers) - - if response.status_code == 200: - records = response.json() - if not records: - print("No transactions exceeding total balances found.") - else: - print("\nTransactions Exceeding Total Balances:") - for record in records: - print("-" * 50) - print(f"User ID: {record['user_id']}") - print(f"User Name: {record['user_name']}") - print(f"Transaction ID: {record['transaction_id']}") - print(f"Transaction Name: {record['transaction_name']}") - print(f"Transaction Amount: {record['transaction_amount']:,.2f}") - print(f"Transaction Date: {record['transaction_date']}") - print(f"Calculated Total Balance: {record['calculated_total_balance']:,.2f}") - print("-" * 50) - else: - print("Failed to fetch the report.") - error_message = response.json().get("detail", "Unknown error") - print(f"Error: {error_message}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - -def get_exceeding_user_total_balance(): - headers = {"Authorization": f"Bearer {access_token}"} - - print("\nTransactions Exceeding User's Total Balance") - try: - response = requests.get(f"{BASE_URL}/reports/exceeding-user-total-balance", headers=headers) - - if response.status_code == 200: - records = response.json() - if not records: - print("No users with transactions exceeding their total balance.") - else: - print("\nUsers with Transactions Exceeding Total Balance:") - for record in records: - print("-" * 50) - print(f"User ID: {record['user_id']}") - print(f"User Name: {record['user_name']}") - print(f"Total Transaction Amount: {record['total_transaction_amount']:,.2f}") - print(f"Total Balance: {record['user_total_balance']:,.2f}") - print("-" * 50) - else: - print("Failed to fetch the report.") - error_message = response.json().get("detail", "Unknown error") - print(f"Error: {error_message}") - except requests.exceptions.RequestException as e: - print(f"Request failed: {e}") - - -if __name__ == "__main__": - main_menu() - Index: prototype_cli_app/cli_app.py =================================================================== --- prototype_cli_app/cli_app.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) +++ prototype_cli_app/cli_app.py (revision aef0c04d03c946bf83af810d29b48f44d57ebdee) @@ -0,0 +1,640 @@ +import requests +import jwt +from backend_python.auth import is_admin, decode_access_token + +BASE_URL = "http://localhost:8000" +access_token = None # Global variable for the access token + +def main_menu(): + """Displays the main menu for registration and login.""" + global access_token + while True: + print("\nMain Menu") + print("1. Register") + print("2. Login") + print("3. Exit") + choice = input("Choose an option: ") + + if choice == "1": + register() + elif choice == "2": + token = login() + if token: + access_token = token + handle_menu_after_login() + elif choice == "3": + print("Exiting...") + break + else: + print("Invalid choice. Please try again.") + +def register(): + """Handles user registration.""" + print("\nRegister") + user_name = input("Enter your username: ") + email = input("Enter your email: ") + password = input("Enter your password: ") + + response = requests.post(f"{BASE_URL}/auth/register/", json={ + "user_name": user_name, + "email": email, + "password": password + }) + + if response.status_code == 200: + print("Registration successful.") + else: + print("Registration failed. Please try again.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def login(): + """Handles user login and returns the access token.""" + print("\nLog in") + email = input("Enter your email: ") + password = input("Enter your password: ") + + response = requests.post(f"{BASE_URL}/auth/login/", json={ + "email": email, + "password": password + }) + + if response.status_code == 200: + data = response.json() + print(f"Log in successful. Access Token: {data['access_token']}") + return data["access_token"] + else: + print("Invalid login credentials.") + return None + +def handle_menu_after_login(): + """Routes the user to the appropriate menu based on their role.""" + global access_token + # Decode the JWT to extract user information + try: + payload = decode_access_token(access_token) + user_id = payload.get("sub") # Extract user_id from the token + email = payload.get("email") # Extract email from the token + + if not user_id or not email: + raise ValueError("Token is missing required fields.") + + print(f"User ID from token: {user_id}, Email: {email}") + + # Check if the user is an admin + if is_admin(email): + admin_menu() + else: + user_menu() + except jwt.ExpiredSignatureError: + print("Session expired. Please log in again.") + access_token = None + except ValueError as e: + print(f"Token validation error: {str(e)}. Logging out.") + access_token = None + except Exception as e: + print(f"An error occurred: {str(e)}. Logging out.") + access_token = None + +def admin_menu(): + """Displays the admin menu.""" + global access_token + while True: + print("\nAdmin Menu") + print("1. View All Transaction Accounts") + print("2. Log Out") + choice = input("Choose an option: ") + + if choice == "1": + admin_view_all_accounts() + elif choice == "2": + print("Logging out...") + access_token = None + break + else: + print("Invalid choice. Please try again.") + +def admin_view_all_accounts(): + """Fetches and displays all transaction accounts for admin users.""" + headers = {"Authorization": f"Bearer {access_token}"} + print("\nAll Transaction Accounts") + response = requests.get(f"{BASE_URL}/admin/accounts/", headers=headers) + + if response.status_code == 200: + accounts = response.json() + for account in accounts: + print(f"Account ID: {account['transaction_account_id']}\n" + +f" Name: {account['account_name']}\n" + +f" Balance: {account['balance']}\n") + else: + print("Failed to retrieve accounts.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def user_menu(): + """Displays the user menu.""" + global access_token + while True: + print("\nUser Menu") + print("1. Add Transaction Account") + print("2. View Transaction Accounts") + print("3. Add Transaction") + print("4. View Transactions") + print("5. Modify Transaction") + print("6. Delete Transaction") + print("7. Add Tag") + print("8. View Tags") + print("9. Assign Tag to Transaction") + print("10. View Transaction Tags") + print("11. View Reports") + print("12. Log Out") + choice = input("Choose an option: ") + + if choice == "1": + add_transaction_account() + elif choice == "2": + view_transaction_accounts() + elif choice == "3": + add_transaction() + elif choice == "4": + view_transactions() + elif choice == "5": + modify_transaction() + elif choice == "6": + delete_transaction() + elif choice == "7": + add_tag() + elif choice == "8": + view_tags() + elif choice == "9": + assign_tag_to_transaction() + elif choice == "10": + view_transaction_tags() + elif choice == "11": + reports_menu() + elif choice == "12": + print("Logging out...") + access_token = None + break + else: + print("Invalid choice. Please try again.") + +def add_transaction_account(): + headers = {"Authorization": f"Bearer {access_token}"} + print("\nAdd Transaction Account") + account_name = input("Enter account name: ") + balance = float(input("Enter balance: ")) + + response = requests.post(f"{BASE_URL}/accounts/", json={ + "account_name": account_name, + "balance": balance + }, headers=headers) + + if response.status_code == 200: + print("Transaction account added successfully.") + else: + print("Failed to add transaction account.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def view_transaction_accounts(): + headers = {"Authorization": f"Bearer {access_token}"} + print("\nTransaction Accounts") + response = requests.get(f"{BASE_URL}/accounts/", headers=headers) + + if response.status_code == 200: + accounts = response.json() + for account in accounts: + print(f"Account ID: {account['transaction_account_id']}") + print(f"Account Name: {account['account_name']}") + print(f"Balance: {account['balance']}") + print() + else: + print("Failed to retrieve transaction accounts.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def add_transaction(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nAdd Transaction") + transaction_name = input("Enter transaction name: ") + amount = float(input("Enter amount (default 0): ") or 0) + date = input("Enter date (YYYY-MM-DDTHH:MM:SS+HH:MM: ") + target_account_id = int(input("Enter target transaction account ID: ")) + tag_id = int(input("Enter tag ID (leave blank for none): ") or 0) + + breakdowns = [] + add_breakdown = input("Do you want to add breakdowns? (yes/no): ").lower() + while add_breakdown == "yes": + breakdown_account_id = int(input("Enter breakdown account ID: ")) + earned_amount = float(input("Enter earned amount (default 0): ") or 0) + spent_amount = float(input("Enter spent amount (default 0): ") or 0) + breakdowns.append({ + "transaction_account_id": breakdown_account_id, + "earned_amount": earned_amount, + "spent_amount": spent_amount + }) + add_breakdown = input("Add another breakdown? (yes/no): ").lower() + + payload = { + "transaction_name": transaction_name, + "amount": amount, + "date": date, + "target_account_id": target_account_id, + "tag_id": tag_id or None, + "breakdowns": breakdowns + } + + response = requests.post(f"{BASE_URL}/transactions/", json=payload, headers=headers) + + if response.status_code == 200: + print("Transaction added successfully.") + else: + print("Failed to add transaction.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def view_transactions(): + headers = {"Authorization": f"Bearer {access_token}"} + print("\nYour Transactions:") + response = requests.get(f"{BASE_URL}/transactions/", headers=headers) + + if response.status_code == 200: + transactions = response.json() + for transaction in transactions: + print(f"Transaction ID: {transaction['transaction_id']}") + print(f" Name: {transaction['transaction_name']}") + print(f" Amount: {transaction['amount']}") + print(f" Net Amount: {transaction['net_amount']}") + print(f" Date: {transaction['date']}") + print(f" Breakdowns:") + + breakdown_response = requests.get( + f"{BASE_URL}/transactions/{transaction['transaction_id']}/breakdowns", + headers=headers + ) + + if breakdown_response.status_code == 200: + breakdowns = breakdown_response.json() + for breakdown in breakdowns: + print(f" Account ID: {breakdown['transaction_account_id']}") + print(f" Earned: {breakdown['earned_amount']}") + print(f" Spent: {breakdown['spent_amount']}") + else: + print(f" Could not fetch breakdowns. Error: {breakdown_response.json().get('detail', 'Unknown error')}") + print() + else: + print("Failed to retrieve transactions.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def modify_transaction(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nModify Transaction") + transaction_id = int(input("Enter transaction ID to modify: ")) + transaction_name = input("Enter new transaction name (or leave blank): ") + amount = input("Enter new amount (or leave blank): ") + net_amount = input("Enter new net amount (or leave blank): ") + date = input("Enter new date (YYYY-MM-DDTHH:MM:SS+HH:MM) (or leave blank): ") + + # Build the request payload + data = {} + if transaction_name: + data["transaction_name"] = transaction_name + if amount: + data["amount"] = float(amount) + if net_amount: + data["net_amount"] = float(net_amount) + if date: + data["date"] = date + + # Make the PUT request with the Authorization header + response = requests.put(f"{BASE_URL}/transactions/{transaction_id}", json=data, headers=headers) + + if response.status_code == 200: + print("Transaction modified successfully.") + else: + print("Failed to modify transaction.") + error_message = response.json().get('detail', 'Unknown error') + print(f"Error: {error_message}") + +def delete_transaction(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nDelete Transaction") + transaction_id = int(input("Enter transaction ID to delete: ")) + + response = requests.delete(f"{BASE_URL}/transactions/{transaction_id}", headers=headers) + + if response.status_code == 200: + print("Transaction deleted successfully.") + else: + print("Failed to delete transaction.") + error_message = response.json().get('detail', 'Unknown error') + print(f"Error: {error_message}") + +def add_tag(): + """Handles creating a new tag for the logged-in user.""" + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nAdd Tag") + tag_name = input("Enter tag name: ") + + response = requests.post(f"{BASE_URL}/tags/", json={"tag_name": tag_name}, headers=headers) + + if response.status_code == 200: + print("Tag added successfully.") + else: + print("Failed to add tag.") + error_message = response.json().get('detail', 'Unknown error') + print(f"Error: {error_message}") + +def view_tags(): + """Retrieve and display tags accessible to the logged-in user.""" + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nYour Tags") + response = requests.get(f"{BASE_URL}/tags/", headers=headers) + + if response.status_code == 200: + tags = response.json() + for tag in tags: + print(f"ID: {tag['tag_id']}, Name: {tag['tag_name']}") + else: + print(f"Failed to retrieve tags. Error: {response.json().get('detail', 'Unknown error')}") + +def assign_tag_to_transaction(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nAssign Tag to Transaction") + transaction_id = int(input("Enter transaction ID: ")) + tag_id = int(input("Enter tag ID: ")) + + response = requests.post(f"{BASE_URL}/tags/assign/", json={ + "transaction_id": transaction_id, + "tag_id": tag_id + }, headers=headers) + + if response.status_code == 200: + print("Tag assigned successfully.") + else: + print("Failed to assign tag.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + +def view_transaction_tags(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nView Transaction Tags") + transaction_id = int(input("Enter transaction ID to view its tags: ")) + + response = requests.get(f"{BASE_URL}/transactions/{transaction_id}", headers=headers) + + if response.status_code == 200: + transaction = response.json() + print(f"Transaction: {transaction['transaction_name']}") + + # Fetch tags linked to the transaction + tags_response = requests.get(f"{BASE_URL}/tags/transaction/{transaction_id}", headers=headers) + if tags_response.status_code == 200: + tags = tags_response.json() + for tag in tags: + print(f"Tag ID: {tag['tag_id']}, Name: {tag['tag_name']}") + else: + print(f"Failed to retrieve tags. Error: {tags_response.json().get('detail', 'Unknown error')}") + else: + print(f"Failed to retrieve transaction. Error: {response.json().get('detail', 'Unknown error')}") + +def reports_menu(): + """ + Displays the reports menu. + """ + global access_token + + while True: + print("\nReports Menu") + print("0. Go back to User Menu") + print("1. View Total Spending") + print("2. View Spending by Category") + print("3. View Spending by Date Range") + print("4. View Transactions Exceeding Account Balance") + print("5. View Transactions Exceeding Account Balance Currently") + print("6. View Chronological List of Transactions Exceeding Total Account Balances") + print("7. View Transactions Exceeding User's Total Current Balance") + choice = input("Choose a report option: ") + + if choice == "1": + get_total_spending() + elif choice == "2": + get_spending_by_category() + elif choice == "3": + get_spending_by_date_range() + elif choice == "4": + get_exceeding_account_balance() + elif choice == "5": + get_exceeding_current_balance() + elif choice == "6": + get_exceeding_total_balances() + elif choice == "7": + get_exceeding_user_total_balance() + elif choice == "0": + break + else: + print("Invalid choice. Please try again.") + +def get_total_spending(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nTotal Spending Report") + try: + response = requests.get(f"{BASE_URL}/reports/total-spending", headers=headers) + + if response.status_code == 200: + data = response.json() + total_spent = data.get("total_spent", 0.0) + print(f"Your total spending is: {float(total_spent):,.2f}") + else: + print("Failed to fetch the total spending report.") + error_message = response.json().get("detail", "Unknown error") + print(f"Error: {error_message}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + +def get_spending_by_category(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nSpending by Category Report") + try: + response = requests.get(f"{BASE_URL}/reports/spending-by-category", headers=headers) + + if response.status_code == 200: + data = response.json() + spending_by_category = data.get("spending_by_category", {}) + + if not spending_by_category: + print("No spending data found by category.") + else: + print("\nSpending by Category:") + print("-" * 50) + for category, amount in spending_by_category.items(): + print(f"Category: {category}") + print(f"Total Spending: {float(amount):,.2f}") + print("-" * 50) + else: + print("Failed to fetch the spending by category report.") + error_message = response.json().get("detail", "Unknown error") + print(f"Error: {error_message}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + +def get_spending_by_date_range(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nSpending by Date Range Report") + start_date = input("Enter start date (YYYY-MM-DD): ") + end_date = input("Enter end date (YYYY-MM-DD): ") + + try: + response = requests.get( + f"{BASE_URL}/reports/spending-by-date-range", + params={"start_date": start_date, "end_date": end_date}, + headers=headers + ) + + if response.status_code == 200: + data = response.json() + total_spent = data.get("total_spent", 0.0) + print("\nSpending by Date Range:") + print(f"From: {start_date} To: {end_date}") + print(f"Total Spending: {float(total_spent):,.2f}") + else: + print("Failed to fetch the spending by date range report.") + error_message = response.json().get("detail", "Unknown error") + print(f"Error: {error_message}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + +def get_exceeding_account_balance(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nTransactions Exceeding Account Balance Report") + account_name = input("Enter account name (leave blank for all accounts): ") + params = {"account_name": account_name} if account_name else {} + + try: + response = requests.get( + f"{BASE_URL}/reports/exceeding-transactions", + params=params, + headers=headers + ) + + if response.status_code == 200: + records = response.json() + if not records: + print("No transactions exceeding account balance were found.") + else: + print("\nExceeding Transactions:") + for record in records: + print("-" * 50) + print(f"User ID: {record['user_id']}") + print(f"User Name: {record['user_name']}") + print(f"Account Name: {record['account_name']}") + print(f"Transaction ID: {record['transaction_id']}") + print(f"Transaction Name: {record['transaction_name']}") + print(f"Transaction Amount: {float(record['transaction_amount']):,.2f}") + print(f"Transaction Date: {record['transaction_date']}") + print(f"Calculated Balance: {float(record['calculated_balance']):,.2f}") + print("-" * 50) + else: + print("Failed to fetch the exceeding transactions report.") + print(f"Error: {response.json().get('detail', 'Unknown error')}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + +def get_exceeding_current_balance(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nTransactions Exceeding Current Account Balance Report") + try: + response = requests.get(f"{BASE_URL}/reports/exceeding-current-balance", headers=headers) + + if response.status_code == 200: + records = response.json() + if not records: + print("No transactions exceeding current account balance were found.") + else: + print("\nTransactions Exceeding Current Balance:") + for record in records: + print("-" * 50) + print(f"User ID: {record['user_id']}") + print(f"User Name: {record['user_name']}") + print(f"Account Name: {record['account_name']}") + print(f"Current Balance: {record['current_balance']:,.2f}") + print(f"Transaction ID: {record['transaction_id']}") + print(f"Transaction Name: {record['transaction_name']}") + print(f"Transaction Amount: {record['transaction_amount']:,.2f}") + print(f"Transaction Date: {record['transaction_date']}") + print("-" * 50) + else: + print("Failed to fetch the report.") + error_message = response.json().get("detail", "Unknown error") + print(f"Error: {error_message}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + +def get_exceeding_total_balances(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nChronological List of Transactions Exceeding Total Account Balances") + try: + response = requests.get(f"{BASE_URL}/reports/exceeding-total-balances", headers=headers) + + if response.status_code == 200: + records = response.json() + if not records: + print("No transactions exceeding total balances found.") + else: + print("\nTransactions Exceeding Total Balances:") + for record in records: + print("-" * 50) + print(f"User ID: {record['user_id']}") + print(f"User Name: {record['user_name']}") + print(f"Transaction ID: {record['transaction_id']}") + print(f"Transaction Name: {record['transaction_name']}") + print(f"Transaction Amount: {record['transaction_amount']:,.2f}") + print(f"Transaction Date: {record['transaction_date']}") + print(f"Calculated Total Balance: {record['calculated_total_balance']:,.2f}") + print("-" * 50) + else: + print("Failed to fetch the report.") + error_message = response.json().get("detail", "Unknown error") + print(f"Error: {error_message}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + +def get_exceeding_user_total_balance(): + headers = {"Authorization": f"Bearer {access_token}"} + + print("\nTransactions Exceeding User's Total Balance") + try: + response = requests.get(f"{BASE_URL}/reports/exceeding-user-total-balance", headers=headers) + + if response.status_code == 200: + records = response.json() + if not records: + print("No users with transactions exceeding their total balance.") + else: + print("\nUsers with Transactions Exceeding Total Balance:") + for record in records: + print("-" * 50) + print(f"User ID: {record['user_id']}") + print(f"User Name: {record['user_name']}") + print(f"Total Transaction Amount: {record['total_transaction_amount']:,.2f}") + print(f"Total Balance: {record['user_total_balance']:,.2f}") + print("-" * 50) + else: + print("Failed to fetch the report.") + error_message = response.json().get("detail", "Unknown error") + print(f"Error: {error_message}") + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}") + + +if __name__ == "__main__": + main_menu() +