171 lines
4.9 KiB
Python

from typing import List, Optional, Dict, Any
from uuid import UUID
from sqlalchemy.orm import Session
from models.expense import Expense
from schemas.expense import ExpenseCreate, ExpenseUpdate, ExpenseSchema
def get_all_expenses(db: Session) -> List[ExpenseSchema]:
"""
Retrieves all expenses from the database.
Args:
db (Session): The database session.
Returns:
List[ExpenseSchema]: A list of all expense objects.
"""
expenses = db.query(Expense).all()
return [ExpenseSchema.from_orm(expense) for expense in expenses]
def get_expense_by_id(db: Session, expense_id: UUID) -> Optional[ExpenseSchema]:
"""
Retrieves a single expense by its ID.
Args:
db (Session): The database session.
expense_id (UUID): The ID of the expense to retrieve.
Returns:
Optional[ExpenseSchema]: The expense object if found, otherwise None.
"""
expense = db.query(Expense).filter(Expense.id == expense_id).first()
return ExpenseSchema.from_orm(expense) if expense else None
def create_expense(db: Session, expense_data: ExpenseCreate) -> ExpenseSchema:
"""
Creates a new expense in the database.
Args:
db (Session): The database session.
expense_data (ExpenseCreate): The data for the expense to create.
Returns:
ExpenseSchema: The newly created expense object.
"""
db_expense = Expense(**expense_data.dict())
db.add(db_expense)
db.commit()
db.refresh(db_expense)
return ExpenseSchema.from_orm(db_expense)
def update_expense(db: Session, expense_id: UUID, expense_data: ExpenseUpdate) -> Optional[ExpenseSchema]:
"""
Updates an existing expense in the database.
Args:
db (Session): The database session.
expense_id (UUID): The ID of the expense to update.
expense_data (ExpenseUpdate): The updated data for the expense.
Returns:
Optional[ExpenseSchema]: The updated expense object if found, otherwise None.
"""
expense = db.query(Expense).filter(Expense.id == expense_id).first()
if not expense:
return None
expense_data = expense_data.dict(exclude_unset=True)
for key, value in expense_data.items():
setattr(expense, key, value)
db.add(expense)
db.commit()
db.refresh(expense)
return ExpenseSchema.from_orm(expense)
def delete_expense(db: Session, expense_id: UUID) -> bool:
"""
Deletes an expense from the database.
Args:
db (Session): The database session.
expense_id (UUID): The ID of the expense to delete.
Returns:
bool: True if the expense was successfully deleted, False otherwise.
"""
expense = db.query(Expense).filter(Expense.id == expense_id).first()
if not expense:
return False
db.delete(expense)
db.commit()
return True
def validate_expense_data(expense_data: Dict[str, Any]) -> bool:
"""
Validates the expense data dictionary.
Args:
expense_data (Dict[str, Any]): The expense data to validate.
Returns:
bool: True if the data is valid, False otherwise.
"""
if not expense_data:
return False
if "title" not in expense_data or not isinstance(expense_data["title"], str) or len(expense_data["title"]) < 3:
return False
if "amount" not in expense_data or not isinstance(expense_data["amount"], (int, float)) or expense_data["amount"] <= 0:
return False
if "description" in expense_data and not isinstance(expense_data["description"], str):
return False
if "category" in expense_data and not isinstance(expense_data["category"], str):
return False
return True
def validate_expense_title(title: str) -> bool:
"""
Validates the expense title string.
Args:
title (str): The expense title to validate.
Returns:
bool: True if the title is valid, False otherwise.
"""
if not isinstance(title, str) or len(title) < 3:
return False
# Add additional validation rules for the title if needed
return True
def validate_expense_amount(amount: float) -> bool:
"""
Validates the expense amount.
Args:
amount (float): The expense amount to validate.
Returns:
bool: True if the amount is valid, False otherwise.
"""
if not isinstance(amount, (int, float)) or amount <= 0:
return False
# Add additional validation rules for the amount if needed
return True
def validate_expense_category(category: Optional[str]) -> bool:
"""
Validates the expense category string.
Args:
category (Optional[str]): The expense category to validate.
Returns:
bool: True if the category is valid, False otherwise.
"""
if category is None:
return True
if not isinstance(category, str):
return False
# Add additional validation rules for the category if needed
return True