171 lines
4.9 KiB
Python
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 |