Build a personal expense management system with FastAPI backend. The system will manage expenses, categories, and budgets with proper validation and reporting capabilities. We'll focus on Repository Pattern implementation with comprehensive testing at each layer.
Pattern: Build → Test → Move Forward
Learning Objectives:
Tasks:
python -m venv expense_env
source expense_env/bin/activate # Linux/Mac
pip install fastapi uvicorn sqlalchemy python-multipart pytest pytest-asyncio # main.py
from fastapi import FastAPI
app = FastAPI(title="Expense Management API", version="1.0.0")
@app.get("/")
def read_root():
return {"message": "Welcome to Expense Management System"}
@app.get("/health")
def health_check():
return {"status": "healthy"}Deliverable: Running FastAPI application with basic endpoints
Learning Objectives:
Tasks:
# models/base.py
from sqlalchemy import Column, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class TimestampMixin:
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# models/user.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from .base import Base, TimestampMixin
class User(Base, TimestampMixin):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
full_name = Column(String)
department = Column(String)
expenses = relationship("Expense", back_populates="user")
budgets = relationship("Budget", back_populates="user")
# models/category.py
from sqlalchemy import Column, Integer, String, Text
from sqlalchemy.orm import relationship
from .base import Base, TimestampMixin
class Category(Base, TimestampMixin):
__tablename__ = "categories"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, unique=True, index=True)
description = Column(Text)
expenses = relationship("Expense", back_populates="category")
budgets = relationship("Budget", back_populates="category")
# models/expense.py
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship
from .base import Base, TimestampMixin
class Expense(Base, TimestampMixin):
__tablename__ = "expenses"
id = Column(Integer, primary_key=True, index=True)
amount = Column(Float, nullable=False)
description = Column(Text)
expense_date = Column(DateTime, nullable=False)
merchant_name = Column(String) # From payload instead of OCR
receipt_number = Column(String) # From payload instead of OCR
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
category_id = Column(Integer, ForeignKey("categories.id"), nullable=False)
user = relationship("User", back_populates="expenses")
category = relationship("Category", back_populates="expenses")
# models/budget.py
from sqlalchemy import Column, Integer, Float, ForeignKey
from sqlalchemy.orm import relationship
from .base import Base, TimestampMixin
class Budget(Base, TimestampMixin):
__tablename__ = "budgets"
id = Column(Integer, primary_key=True, index=True)
amount = Column(Float, nullable=False)
month = Column(Integer, nullable=False) # 1-12
year = Column(Integer, nullable=False)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
category_id = Column(Integer, ForeignKey("categories.id"), nullable=False)
user = relationship("User", back_populates="budgets")
category = relationship("Category", back_populates="budgets") # core/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./expense_management.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Create tables
from models.base import Base
Base.metadata.create_all(bind=engine)Deliverable: Complete database models with relationships
Learning Objectives:
Tasks:
# repositories/base.py
from typing import Generic, TypeVar, Type, Optional, List, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import DeclarativeMeta
ModelType = TypeVar("ModelType", bound=DeclarativeMeta)
class BaseRepository(Generic[ModelType]):
def __init__(self, model: Type[ModelType], db: Session):
self.model = model
self.db = db
def get(self, id: int) -> Optional[ModelType]:
return self.db.query(self.model).filter(self.model.id == id).first()
def get_all(self, skip: int = 0, limit: int = 100) -> List[ModelType]:
return self.db.query(self.model).offset(skip).limit(limit).all()
def create(self, obj_in: Dict[str, Any]) -> ModelType:
db_obj = self.model(**obj_in)
self.db.add(db_obj)
self.db.commit()
self.db.refresh(db_obj)
return db_obj
def update(self, db_obj: ModelType, obj_in: Dict[str, Any]) -> ModelType:
for field, value in obj_in.items():
if hasattr(db_obj, field):
setattr(db_obj, field, value)
self.db.commit()
self.db.refresh(db_obj)
return db_obj
def delete(self, id: int) -> bool:
obj = self.get(id)
if obj:
self.db.delete(obj)
self.db.commit()
return True
return False
def count(self) -> int:
return self.db.query(self.model).count() # repositories/user.py
from .base import BaseRepository
from models.user import User
from typing import Optional
class UserRepository(BaseRepository[User]):
def get_by_email(self, email: str) -> Optional[User]:
return self.db.query(User).filter(User.email == email).first()
def get_by_department(self, department: str) -> List[User]:
return self.db.query(User).filter(User.department == department).all() # repositories/category.py
from .base import BaseRepository
from models.category import Category
from typing import Optional, List
class CategoryRepository(BaseRepository[Category]):
def get_by_name(self, name: str) -> Optional[Category]:
return self.db.query(Category).filter(Category.name == name).first()
def search_by_name(self, search_term: str) -> List[Category]:
return self.db.query(Category).filter(
Category.name.contains(search_term)
).all() # repositories/expense.py
from .base import BaseRepository
from models.expense import Expense
from typing import List, Optional
from datetime import datetime, date
from sqlalchemy import func, extract, and_, desc
class ExpenseRepository(BaseRepository[Expense]):
def get_by_user(self, user_id: int, skip: int = 0, limit: int = 100) -> List[Expense]:
return (self.db.query(Expense)
.filter(Expense.user_id == user_id)
.order_by(desc(Expense.expense_date))
.offset(skip)
.limit(limit)
.all())
def get_by_category(self, category_id: int, user_id: Optional[int] = None) -> List[Expense]:
query = self.db.query(Expense).filter(Expense.category_id == category_id)
if user_id:
query = query.filter(Expense.user_id == user_id)
return query.all()
def get_by_date_range(self, start_date: date, end_date: date, user_id: Optional[int] = None) -> List[Expense]:
query = self.db.query(Expense).filter(
and_(
Expense.expense_date >= start_date,
Expense.expense_date <= end_date
)
)
if user_id:
query = query.filter(Expense.user_id == user_id)
return query.order_by(desc(Expense.expense_date)).all()
def get_monthly_total(self, user_id: int, month: int, year: int) -> float:
result = (self.db.query(func.sum(Expense.amount))
.filter(
Expense.user_id == user_id,
extract('month', Expense.expense_date) == month,
extract('year', Expense.expense_date) == year
).scalar())
return result or 0.0
def get_category_total(self, user_id: int, category_id: int, month: int, year: int) -> float:
result = (self.db.query(func.sum(Expense.amount))
.filter(
Expense.user_id == user_id,
Expense.category_id == category_id,
extract('month', Expense.expense_date) == month,
extract('year', Expense.expense_date) == year
).scalar())
return result or 0.0
def get_total_by_merchant(self, user_id: int, merchant_name: str) -> float:
result = (self.db.query(func.sum(Expense.amount))
.filter(
Expense.user_id == user_id,
Expense.merchant_name == merchant_name
).scalar())
return result or 0.0
def search_by_description(self, user_id: int, search_term: str) -> List[Expense]:
return (self.db.query(Expense)
.filter(
Expense.user_id == user_id,
Expense.description.contains(search_term)
).all()) # repositories/budget.py
from .base import BaseRepository
from models.budget import Budget
from typing import List, Optional
class BudgetRepository(BaseRepository[Budget]):
def get_by_user(self, user_id: int) -> List[Budget]:
return self.db.query(Budget).filter(Budget.user_id == user_id).all()
def get_by_period(self, user_id: int, month: int, year: int) -> List[Budget]:
return (self.db.query(Budget)
.filter(
Budget.user_id == user_id,
Budget.month == month,
Budget.year == year
).all())
def get_monthly_budget(self, user_id: int, category_id: int, month: int, year: int) -> Optional[Budget]:
return (self.db.query(Budget)
.filter(
Budget.user_id == user_id,
Budget.category_id == category_id,
Budget.month == month,
Budget.year == year
).first())
def get_yearly_budgets(self, user_id: int, year: int) -> List[Budget]:
return (self.db.query(Budget)
.filter(
Budget.user_id == user_id,
Budget.year == year
).all())Learning Objectives:
Tasks:
# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models.base import Base
from models.user import User
from models.category import Category
from models.expense import Expense
from models.budget import Budget
from datetime import datetime, date
# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="function")
def db_session():
# Create tables
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
# Drop tables after test
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def sample_user(db_session):
user_data = {
"email": "test@example.com",
"full_name": "Test User",
"department": "Engineering"
}
user = User(**user_data)
db_session.add(user)
db_session.commit()
db_session.refresh(user)
return user
@pytest.fixture
def sample_category(db_session):
category_data = {
"name": "Food & Dining",
"description": "Restaurant and food expenses"
}
category = Category(**category_data)
db_session.add(category)
db_session.commit()
db_session.refresh(category)
return category
@pytest.fixture
def sample_expense(db_session, sample_user, sample_category):
expense_data = {
"amount": 25.50,
"description": "Lunch at restaurant",
"expense_date": datetime(2024, 3, 15, 12, 30),
"merchant_name": "Pizza Palace",
"receipt_number": "R12345",
"user_id": sample_user.id,
"category_id": sample_category.id
}
expense = Expense(**expense_data)
db_session.add(expense)
db_session.commit()
db_session.refresh(expense)
return expense # tests/test_repositories/test_expense_repository.py
import pytest
from repositories.expense import ExpenseRepository
from models.expense import Expense
from datetime import datetime, date
class TestExpenseRepository:
def test_create_expense(self, db_session, sample_user, sample_category):
"""Test creating a new expense"""
repo = ExpenseRepository(Expense, db_session)
expense_data = {
"amount": 99.99,
"description": "Business lunch",
"expense_date": datetime(2024, 3, 15),
"merchant_name": "Restaurant ABC",
"receipt_number": "REC001",
"user_id": sample_user.id,
"category_id": sample_category.id
}
expense = repo.create(expense_data)
assert expense.id is not None
assert expense.amount == 99.99
assert expense.description == "Business lunch"
assert expense.merchant_name == "Restaurant ABC"
assert expense.user_id == sample_user.id
def test_get_expense_by_id(self, db_session, sample_expense):
"""Test retrieving expense by ID"""
repo = ExpenseRepository(Expense, db_session)
found_expense = repo.get(sample_expense.id)
assert found_expense is not None
assert found_expense.id == sample_expense.id
assert found_expense.amount == sample_expense.amount
def test_get_by_user(self, db_session, sample_user, sample_category):
"""Test retrieving expenses by user"""
repo = ExpenseRepository(Expense, db_session)
# Create multiple expenses for the user
for i in range(3):
expense_data = {
"amount": 10.0 + i,
"description": f"Expense {i}",
"expense_date": datetime(2024, 3, 15 + i),
"merchant_name": f"Store {i}",
"user_id": sample_user.id,
"category_id": sample_category.id
}
repo.create(expense_data)
expenses = repo.get_by_user(sample_user.id)
assert len(expenses) == 3
assert all(exp.user_id == sample_user.id for exp in expenses)
def test_get_by_date_range(self, db_session, sample_user, sample_category):
"""Test retrieving expenses by date range"""
repo = ExpenseRepository(Expense, db_session)
# Create expenses with different dates
dates = [
datetime(2024, 3, 10),
datetime(2024, 3, 15),
datetime(2024, 3, 20),
datetime(2024, 4, 1) # Outside range
]
for i, expense_date in enumerate(dates):
expense_data = {
"amount": 10.0 + i,
"description": f"Expense {i}",
"expense_date": expense_date,
"user_id": sample_user.id,
"category_id": sample_category.id
}
repo.create(expense_data)
# Test date range query
start_date = date(2024, 3, 1)
end_date = date(2024, 3, 31)
expenses = repo.get_by_date_range(start_date, end_date, sample_user.id)
assert len(expenses) == 3 # Should exclude April expense
for expense in expenses:
assert start_date <= expense.expense_date.date() <= end_date
def test_get_monthly_total(self, db_session, sample_user, sample_category):
"""Test calculating monthly expense total"""
repo = ExpenseRepository(Expense, db_session)
# Create expenses for March 2024
march_expenses = [
{"amount": 100.0, "expense_date": datetime(2024, 3, 5)},
{"amount": 200.0, "expense_date": datetime(2024, 3, 15)},
{"amount": 150.0, "expense_date": datetime(2024, 4, 1)}, # Different month
]
for expense_data in march_expenses:
expense_data.update({
"description": "Test expense",
"user_id": sample_user.id,
"category_id": sample_category.id
})
repo.create(expense_data)
total = repo.get_monthly_total(sample_user.id, 3, 2024)
assert total == 300.0 # Only March expenses
def test_get_category_total(self, db_session, sample_user):
"""Test calculating category-wise total"""
repo = ExpenseRepository(Expense, db_session)
# Create two categories
category1 = Category(name="Food", description="Food expenses")
category2 = Category(name="Travel", description="Travel expenses")
db_session.add_all([category1, category2])
db_session.commit()
# Create expenses for different categories
expenses_data = [
{"amount": 50.0, "category_id": category1.id},
{"amount": 75.0, "category_id": category1.id},
{"amount": 200.0, "category_id": category2.id},
]
for expense_data in expenses_data:
expense_data.update({
"description": "Test expense",
"expense_date": datetime(2024, 3, 15),
"user_id": sample_user.id
})
repo.create(expense_data)
food_total = repo.get_category_total(sample_user.id, category1.id, 3, 2024)
travel_total = repo.get_category_total(sample_user.id, category2.id, 3, 2024)
assert food_total == 125.0
assert travel_total == 200.0
def test_update_expense(self, db_session, sample_expense):
"""Test updating an expense"""
repo = ExpenseRepository(Expense, db_session)
update_data = {
"amount": 150.0,
"description": "Updated description"
}
updated_expense = repo.update(sample_expense, update_data)
assert updated_expense.amount == 150.0
assert updated_expense.description == "Updated description"
assert updated_expense.id == sample_expense.id
def test_delete_expense(self, db_session, sample_expense):
"""Test deleting an expense"""
repo = ExpenseRepository(Expense, db_session)
expense_id = sample_expense.id
result = repo.delete(expense_id)
assert result is True
assert repo.get(expense_id) is None
def test_search_by_description(self, db_session, sample_user, sample_category):
"""Test searching expenses by description"""
repo = ExpenseRepository(Expense, db_session)
expenses_data = [
{"description": "Coffee at Starbucks"},
{"description": "Lunch meeting with client"},
{"description": "Office supplies purchase"},
]
for expense_data in expenses_data:
expense_data.update({
"amount": 25.0,
"expense_date": datetime(2024, 3, 15),
"user_id": sample_user.id,
"category_id": sample_category.id
})
repo.create(expense_data)
results = repo.search_by_description(sample_user.id, "Coffee")
assert len(results) == 1
assert "Coffee" in results[0].description
# tests/test_repositories/test_budget_repository.py
import pytest
from repositories.budget import BudgetRepository
from models.budget import Budget
class TestBudgetRepository:
def test_create_budget(self, db_session, sample_user, sample_category):
"""Test creating a new budget"""
repo = BudgetRepository(Budget, db_session)
budget_data = {
"amount": 500.0,
"month": 3,
"year": 2024,
"user_id": sample_user.id,
"category_id": sample_category.id
}
budget = repo.create(budget_data)
assert budget.id is not None
assert budget.amount == 500.0
assert budget.month == 3
assert budget.year == 2024
def test_get_monthly_budget(self, db_session, sample_user, sample_category):
"""Test retrieving monthly budget for specific category"""
repo = BudgetRepository(Budget, db_session)
budget_data = {
"amount": 300.0,
"month": 3,
"year": 2024,
"user_id": sample_user.id,
"category_id": sample_category.id
}
created_budget = repo.create(budget_data)
found_budget = repo.get_monthly_budget(
sample_user.id, sample_category.id, 3, 2024
)
assert found_budget is not None
assert found_budget.id == created_budget.id
assert found_budget.amount == 300.0
def test_get_by_period(self, db_session, sample_user):
"""Test retrieving all budgets for a specific period"""
repo = BudgetRepository(Budget, db_session)
# Create multiple categories and budgets
categories = []
for i in range(3):
category = Category(name=f"Category {i}", description=f"Category {i}")
db_session.add(category)
db_session.commit()
categories.append(category)
# Create budgets for March 2024
for category in categories:
budget_data = {
"amount": 100.0 + (category.id * 50),
"month": 3,
"year": 2024,
"user_id": sample_user.id,
"category_id": category.id
}
repo.create(budget_data)
budgets = repo.get_by_period(sample_user.id, 3, 2024)
assert len(budgets) == 3
assert all(b.month == 3 and b.year == 2024 for b in budgets)Deliverable: Complete repository layer with comprehensive tests
Learning Objectives:
Tasks:
# schemas/expense.py
from pydantic import BaseModel, validator
from datetime import datetime
from typing import Optional
class ExpenseBase(BaseModel):
amount: float
description: str
expense_date: datetime
merchant_name: Optional[str] = None
receipt_number: Optional[str] = None
category_id: int
class ExpenseCreate(ExpenseBase):
@validator('amount')
def amount_must_be_positive(cls, v):
if v <= 0:
raise ValueError('Amount must be positive')
return v
class ExpenseUpdate(BaseModel):
amount: Optional[float] = None
description: Optional[str] = None
expense_date: Optional[datetime] = None
merchant_name: Optional[str] = None
receipt_number: Optional[str] = None
category_id: Optional[int] = None
class ExpenseResponse(ExpenseBase):
id: int
user_id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
# schemas/budget.py
from pydantic import BaseModel, validator
from typing import Optional
class BudgetBase(BaseModel):
amount: float
month: int
year: int
category_id: int
class BudgetCreate(BudgetBase):
@validator('amount')
def amount_must_be_positive(cls, v):
if v <= 0:
raise ValueError('Budget amount must be positive')
return v
@validator('month')
def month_must_be_valid(cls, v):
if not 1 <= v <= 12:
raise ValueError('Month must be between 1 and 12')
return v
class BudgetResponse(BudgetBase):
id: int
user_id: int
class Config:
from_attributes = True # services/expense_service.py
from repositories.expense import ExpenseRepository
from repositories.budget import BudgetRepository
from repositories.user import UserRepository
from repositories.category import CategoryRepository
from models.expense import Expense
from models.budget import Budget
from models.user import User
from models.category import Category
from schemas.expense import ExpenseCreate, ExpenseUpdate, ExpenseResponse
from sqlalchemy.orm import Session
from fastapi import HTTPException, status
from typing import List, Optional, Dict
from datetime import datetime, date
class ExpenseService:
def __init__(self, db: Session):
self.db = db
self.expense_repo = ExpenseRepository(Expense, db)
self.budget_repo = BudgetRepository(Budget, db)
self.user_repo = UserRepository(User, db)
self.category_repo = CategoryRepository(Category, db)
def create_expense(self, expense_data: ExpenseCreate, user_id: int) -> Expense:
# Validate user exists
user = self.user_repo.get(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
# Validate category exists
category = self.category_repo.get(expense_data.category_id)
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found"
)
# Check budget limits
self._check_budget_limit(expense_data.category_id, user_id, expense_data.amount)
# Create expense
expense_dict = expense_data.dict()
expense_dict["user_id"] = user_id
return self.expense_repo.create(expense_dict)
def _check_budget_limit(self, category_id: int, user_id: int, amount: float):
current_month = datetime.now().month
current_year = datetime.now().year
# Get monthly budget for category
budget = self.budget_repo.get_monthly_budget(user_id, category_id, current_month, current_year)
if not budget:
return # No budget limit set
# Calculate current month spending
current_spending = self.expense_repo.get_category_total(user_id, category_id, current_month, current_year)
if (current_spending + amount) > budget.amount:
remaining = budget.amount - current_spending
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Expense exceeds monthly budget limit. Remaining: ${remaining:.2f}"
)
def get_expense_by_id(self, expense_id: int, user_id: int) -> Expense:
expense = self.expense_repo.get(expense_id)
if not expense:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Expense not found"
)
# Check ownership
if expense.user_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to view this expense"
)
return expense
def get_user_expenses(
self,
user_id: int,
skip: int = 0,
limit: int = 100,
category_id: Optional[int] = None,
start_date: Optional[date] = None,
end_date: Optional[date] = None
) -> List[Expense]:
if start_date and end_date:
expenses = self.expense_repo.get_by_date_range(start_date, end_date, user_id)
else:
expenses = self.expense_repo.get_by_user(user_id, skip, limit)
# Apply category filter
if category_id:
expenses = [exp for exp in expenses if exp.category_id == category_id]
return expenses
def update_expense(self, expense_id: int, expense_data: ExpenseUpdate, user_id: int) -> Expense:
expense = self.get_expense_by_id(expense_id, user_id)
# If updating category or amount, check budget
update_dict = expense_data.dict(exclude_unset=True)
if 'category_id' in update_dict or 'amount' in update_dict:
new_category_id = update_dict.get('category_id', expense.category_id)
new_amount = update_dict.get('amount', expense.amount)
# Remove old expense amount from budget calculation
current_month = datetime.now().month
current_year = datetime.now().year
current_spending = self.expense_repo.get_category_total(
user_id, expense.category_id, current_month, current_year
)
adjusted_spending = current_spending - expense.amount
# Check new budget with adjusted amount
budget = self.budget_repo.get_monthly_budget(
user_id, new_category_id, current_month, current_year
)
if budget and (adjusted_spending + new_amount) > budget.amount:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Updated expense would exceed budget limit"
)
return self.expense_repo.update(expense, update_dict)
def delete_expense(self, expense_id: int, user_id: int) -> bool:
expense = self.get_expense_by_id(expense_id, user_id)
return self.expense_repo.delete(expense_id)
def get_monthly_summary(self, user_id: int, month: int, year: int) -> Dict:
total_expenses = self.expense_repo.get_monthly_total(user_id, month, year)
expenses = self.expense_repo.get_by_date_range(
date(year, month, 1),
date(year, month + 1, 1) if month < 12 else date(year + 1, 1, 1),
user_id
)
# Group by category
category_totals = {}
for expense in expenses:
if expense.category_id not in category_totals:
category_totals[expense.category_id] = {
'category_name': expense.category.name,
'total': 0,
'count': 0
}
category_totals[expense.category_id]['total'] += expense.amount
category_totals[expense.category_id]['count'] += 1
return {
'month': month,
'year': year,
'total_amount': total_expenses,
'total_expenses': len(expenses),
'category_breakdown': category_totals
} # services/budget_service.py
from repositories.budget import BudgetRepository
from repositories.expense import ExpenseRepository
from repositories.category import CategoryRepository
from models.budget import Budget
from models.expense import Expense
from models.category import Category
from schemas.budget import BudgetCreate, BudgetResponse
from sqlalchemy.orm import Session
from fastapi import HTTPException, status
from typing import List, Dict
from datetime import datetime
class BudgetService:
def __init__(self, db: Session):
self.db = db
self.budget_repo = BudgetRepository(Budget, db)
self.expense_repo = ExpenseRepository(Expense, db)
self.category_repo = CategoryRepository(Category, db)
def create_budget(self, budget_data: BudgetCreate, user_id: int) -> Budget:
# Validate category exists
category = self.category_repo.get(budget_data.category_id)
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found"
)
# Check if budget already exists for the period
existing_budget = self.budget_repo.get_monthly_budget(
user_id, budget_data.category_id, budget_data.month, budget_data.year
)
if existing_budget:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Budget already exists for this category and period"
)
budget_dict = budget_data.dict()
budget_dict["user_id"] = user_id
return self.budget_repo.create(budget_dict)
def get_budget_summary(self, user_id: int, month: int, year: int) -> Dict:
budgets = self.budget_repo.get_by_period(user_id, month, year)
summary = {
"month": month,
"year": year,
"total_budget": 0,
"total_spent": 0,
"categories": []
}
for budget in budgets:
spent = self.expense_repo.get_category_total(
user_id, budget.category_id, month, year
)
category_data = {
"category_id": budget.category_id,
"category_name": budget.category.name,
"budget_amount": budget.amount,
"spent_amount": spent,
"remaining": budget.amount - spent,
"percentage_used": (spent / budget.amount) * 100 if budget.amount > 0 else 0,
"is_over_budget": spent > budget.amount
}
summary["categories"].append(category_data)
summary["total_budget"] += budget.amount
summary["total_spent"] += spent
summary["remaining"] = summary["total_budget"] - summary["total_spent"]
summary["is_over_budget"] = summary["total_spent"] > summary["total_budget"]
return summary
def update_budget(self, budget_id: int, amount: float, user_id: int) -> Budget:
budget = self.budget_repo.get(budget_id)
if not budget:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Budget not found"
)
if budget.user_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to update this budget"
)
return self.budget_repo.update(budget, {"amount": amount})
def delete_budget(self, budget_id: int, user_id: int) -> bool:
budget = self.budget_repo.get(budget_id)
if not budget:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Budget not found"
)
if budget.user_id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to delete this budget"
)
return self.budget_repo.delete(budget_id)Learning Objectives:
Tasks:
# tests/test_services/test_expense_service.py
import pytest
from unittest.mock import Mock, MagicMock
from fastapi import HTTPException
from services.expense_service import ExpenseService
from schemas.expense import ExpenseCreate, ExpenseUpdate
from models.expense import Expense
from models.user import User
from models.category import Category
from models.budget import Budget
from datetime import datetime, date
class TestExpenseService:
@pytest.fixture
def mock_db(self):
return Mock()
@pytest.fixture
def expense_service(self, mock_db):
service = ExpenseService(mock_db)
# Mock repositories
service.expense_repo = Mock()
service.budget_repo = Mock()
service.user_repo = Mock()
service.category_repo = Mock()
return service
@pytest.fixture
def sample_expense_data(self):
return ExpenseCreate(
amount=99.99,
description="Business lunch",
expense_date=datetime(2024, 3, 15),
merchant_name="Restaurant ABC",
receipt_number="REC001",
category_id=1
)
def test_create_expense_success(self, expense_service, sample_expense_data):
"""Test successful expense creation"""
user_id = 1
# Mock repository responses
mock_user = User(id=1, email="test@example.com", full_name="Test User")
mock_category = Category(id=1, name="Food", description="Food expenses")
mock_expense = Expense(id=1, **sample_expense_data.dict(), user_id=user_id)
expense_service.user_repo.get.return_value = mock_user
expense_service.category_repo.get.return_value = mock_category
expense_service.budget_repo.get_monthly_budget.return_value = None # No budget limit
expense_service.expense_repo.create.return_value = mock_expense
result = expense_service.create_expense(sample_expense_data, user_id)
assert result.id == 1
assert result.amount == 99.99
expense_service.expense_repo.create.assert_called_once()
def test_create_expense_user_not_found(self, expense_service, sample_expense_data):
"""Test expense creation with non-existent user"""
user_id = 999
expense_service.user_repo.get.return_value = None
with pytest.raises(HTTPException) as exc_info:
expense_service.create_expense(sample_expense_data, user_id)
assert exc_info.value.status_code == 404
assert "User not found" in str(exc_info.value.detail)
def test_create_expense_category_not_found(self, expense_service, sample_expense_data):
"""Test expense creation with non-existent category"""
user_id = 1
mock_user = User(id=1, email="test@example.com", full_name="Test User")
expense_service.user_repo.get.return_value = mock_user
expense_service.category_repo.get.return_value = None
with pytest.raises(HTTPException) as exc_info:
expense_service.create_expense(sample_expense_data, user_id)
assert exc_info.value.status_code == 404
assert "Category not found" in str(exc_info.value.detail)
def test_create_expense_exceeds_budget(self, expense_service, sample_expense_data):
"""Test expense creation that exceeds budget limit"""
user_id = 1
mock_user = User(id=1, email="test@example.com", full_name="Test User")
mock_category = Category(id=1, name="Food", description="Food expenses")
mock_budget = Budget(amount=50.0, month=3, year=2024) # Budget less than expense
expense_service.user_repo.get.return_value = mock_user
expense_service.category_repo.get.return_value = mock_category
expense_service.budget_repo.get_monthly_budget.return_value = mock_budget
expense_service.expense_repo.get_category_total.return_value = 10.0 # Already spent
with pytest.raises(HTTPException) as exc_info:
expense_service.create_expense(sample_expense_data, user_id)
assert exc_info.value.status_code == 400
assert "exceeds monthly budget limit" in str(exc_info.value.detail)
def test_get_expense_by_id_success(self, expense_service):
"""Test successful expense retrieval"""
expense_id = 1
user_id = 1
mock_expense = Expense(
id=1, amount=99.99, description="Test", user_id=user_id
)
expense_service.expense_repo.get.return_value = mock_expense
result = expense_service.get_expense_by_id(expense_id, user_id)
assert result.id == 1
assert result.user_id == user_id
def test_get_expense_by_id_not_found(self, expense_service):
"""Test expense retrieval with non-existent ID"""
expense_id = 999
user_id = 1
expense_service.expense_repo.get.return_value = None
with pytest.raises(HTTPException) as exc_info:
expense_service.get_expense_by_id(expense_id, user_id)
assert exc_info.value.status_code == 404
assert "Expense not found" in str(exc_info.value.detail)
def test_get_expense_by_id_unauthorized(self, expense_service):
"""Test expense retrieval by unauthorized user"""
expense_id = 1
user_id = 1
mock_expense = Expense(
id=1, amount=99.99, description="Test", user_id=2 # Different user
)
expense_service.expense_repo.get.return_value = mock_expense
with pytest.raises(HTTPException) as exc_info:
expense_service.get_expense_by_id(expense_id, user_id)
assert exc_info.value.status_code == 403
assert "Not authorized" in str(exc_info.value.detail)
def test_update_expense_success(self, expense_service):
"""Test successful expense update"""
expense_id = 1
user_id = 1
mock_expense = Expense(
id=1, amount=99.99, description="Original", user_id=user_id, category_id=1
)
updated_expense = Expense(
id=1, amount=150.0, description="Updated", user_id=user_id, category_id=1
)
expense_service.expense_repo.get.return_value = mock_expense
expense_service.expense_repo.update.return_value = updated_expense
update_data = ExpenseUpdate(amount=150.0, description="Updated")
result = expense_service.update_expense(expense_id, update_data, user_id)
assert result.amount == 150.0
assert result.description == "Updated"
def test_delete_expense_success(self, expense_service):
"""Test successful expense deletion"""
expense_id = 1
user_id = 1
mock_expense = Expense(
id=1, amount=99.99, description="Test", user_id=user_id
)
expense_service.expense_repo.get.return_value = mock_expense
expense_service.expense_repo.delete.return_value = True
result = expense_service.delete_expense(expense_id, user_id)
assert result is True
expense_service.expense_repo.delete.assert_called_once_with(expense_id)
def test_get_monthly_summary(self, expense_service):
"""Test monthly summary generation"""
user_id = 1
month = 3
year = 2024
mock_category = Category(id=1, name="Food")
mock_expenses = [
Expense(id=1, amount=50.0, category_id=1, category=mock_category),
Expense(id=2, amount=75.0, category_id=1, category=mock_category),
]
expense_service.expense_repo.get_monthly_total.return_value = 125.0
expense_service.expense_repo.get_by_date_range.return_value = mock_expenses
result = expense_service.get_monthly_summary(user_id, month, year)
assert result['month'] == month
assert result['year'] == year
assert result['total_amount'] == 125.0
assert result['total_expenses'] == 2
assert 1 in result['category_breakdown']
assert result['category_breakdown'][1]['total'] == 125.0
# tests/test_services/test_budget_service.py
import pytest
from unittest.mock import Mock
from fastapi import HTTPException
from services.budget_service import BudgetService
from schemas.budget import BudgetCreate
from models.budget import Budget
from models.category import Category
class TestBudgetService:
@pytest.fixture
def mock_db(self):
return Mock()
@pytest.fixture
def budget_service(self, mock_db):
service = BudgetService(mock_db)
service.budget_repo = Mock()
service.expense_repo = Mock()
service.category_repo = Mock()
return service
def test_create_budget_success(self, budget_service):
"""Test successful budget creation"""
user_id = 1
budget_data = BudgetCreate(
amount=500.0,
month=3,
year=2024,
category_id=1
)
mock_category = Category(id=1, name="Food")
mock_budget = Budget(id=1, **budget_data.dict(), user_id=user_id)
budget_service.category_repo.get.return_value = mock_category
budget_service.budget_repo.get_monthly_budget.return_value = None
budget_service.budget_repo.create.return_value = mock_budget
result = budget_service.create_budget(budget_data, user_id)
assert result.amount == 500.0
assert result.month == 3
budget_service.budget_repo.create.assert_called_once()
def test_create_budget_category_not_found(self, budget_service):
"""Test budget creation with non-existent category"""
user_id = 1
budget_data = BudgetCreate(
amount=500.0,
month=3,
year=2024,
category_id=999
)
budget_service.category_repo.get.return_value = None
with pytest.raises(HTTPException) as exc_info:
budget_service.create_budget(budget_data, user_id)
assert exc_info.value.status_code == 404
assert "Category not found" in str(exc_info.value.detail)
def test_create_budget_already_exists(self, budget_service):
"""Test budget creation when budget already exists"""
user_id = 1
budget_data = BudgetCreate(
amount=500.0,
month=3,
year=2024,
category_id=1
)
mock_category = Category(id=1, name="Food")
existing_budget = Budget(id=1, amount=300.0, month=3, year=2024)
budget_service.category_repo.get.return_value = mock_category
budget_service.budget_repo.get_monthly_budget.return_value = existing_budget
with pytest.raises(HTTPException) as exc_info:
budget_service.create_budget(budget_data, user_id)
assert exc_info.value.status_code == 400
assert "already exists" in str(exc_info.value.detail)
def test_get_budget_summary(self, budget_service):
"""Test budget summary generation"""
user_id = 1
month = 3
year = 2024
mock_category = Category(id=1, name="Food")
mock_budgets = [
Budget(id=1, amount=500.0, category_id=1, category=mock_category)
]
budget_service.budget_repo.get_by_period.return_value = mock_budgets
budget_service.expense_repo.get_category_total.return_value = 300.0
result = budget_service.get_budget_summary(user_id, month, year)
assert result['month'] == month
assert result['year'] == year
assert result['total_budget'] == 500.0
assert result['total_spent'] == 300.0
assert result['remaining'] == 200.0
assert len(result['categories']) == 1
category_data = result['categories'][0]
assert category_data['budget_amount'] == 500.0
assert category_data['spent_amount'] == 300.0
assert category_data['remaining'] == 200.0
assert category_data['percentage_used'] == 60.0
assert category_data['is_over_budget'] is FalseDeliverable: Complete service layer with comprehensive business logic and tests
Learning Objectives:
Tasks:
# routers/expenses.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List, Optional
from datetime import date
from services.expense_service import ExpenseService
from schemas.expense import ExpenseCreate, ExpenseUpdate, ExpenseResponse
from core.database import get_db
router = APIRouter(prefix="/expenses", tags=["expenses"])
# For simplicity, we'll use a header-based user identification
def get_current_user_id(user_id: int = Depends(lambda: 1)): # Simplified - normally from JWT
return user_id
@router.post("/", response_model=ExpenseResponse, status_code=status.HTTP_201_CREATED)
def create_expense(
expense: ExpenseCreate,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""
Create a new expense entry.
- **amount**: The expense amount (must be positive)
- **description**: Description of the expense
- **expense_date**: When the expense occurred
- **merchant_name**: Name of the merchant (optional)
- **receipt_number**: Receipt number (optional)
- **category_id**: ID of the expense category
"""
service = ExpenseService(db)
return service.create_expense(expense, user_id)
@router.get("/", response_model=List[ExpenseResponse])
def get_expenses(
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, le=1000, description="Maximum number of records to return"),
category_id: Optional[int] = Query(None, description="Filter by category ID"),
start_date: Optional[date] = Query(None, description="Filter expenses from this date"),
end_date: Optional[date] = Query(None, description="Filter expenses until this date"),
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""
Get user's expenses with optional filtering.
- **skip**: Number of records to skip for pagination
- **limit**: Maximum number of records to return
- **category_id**: Filter by specific category
- **start_date**: Filter expenses from this date
- **end_date**: Filter expenses until this date
"""
service = ExpenseService(db)
return service.get_user_expenses(
user_id, skip, limit, category_id, start_date, end_date
)
@router.get("/{expense_id}", response_model=ExpenseResponse)
def get_expense(
expense_id: int,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""Get a specific expense by ID."""
service = ExpenseService(db)
return service.get_expense_by_id(expense_id, user_id)
@router.put("/{expense_id}", response_model=ExpenseResponse)
def update_expense(
expense_id: int,
expense_update: ExpenseUpdate,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""Update an existing expense."""
service = ExpenseService(db)
return service.update_expense(expense_id, expense_update, user_id)
@router.delete("/{expense_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_expense(
expense_id: int,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""Delete an expense."""
service = ExpenseService(db)
service.delete_expense(expense_id, user_id)
return None
@router.get("/summary/monthly")
def get_monthly_summary(
month: int = Query(..., ge=1, le=12, description="Month (1-12)"),
year: int = Query(..., ge=2020, le=2030, description="Year"),
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""Get monthly expense summary with category breakdown."""
service = ExpenseService(db)
return service.get_monthly_summary(user_id, month, year) # routers/budgets.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List
from services.budget_service import BudgetService
from schemas.budget import BudgetCreate, BudgetResponse
from core.database import get_db
router = APIRouter(prefix="/budgets", tags=["budgets"])
def get_current_user_id(user_id: int = Depends(lambda: 1)):
return user_id
@router.post("/", response_model=BudgetResponse, status_code=status.HTTP_201_CREATED)
def create_budget(
budget: BudgetCreate,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""
Create a new budget for a specific category and period.
- **amount**: Budget amount (must be positive)
- **month**: Month (1-12)
- **year**: Year
- **category_id**: ID of the category to budget for
"""
service = BudgetService(db)
return service.create_budget(budget, user_id)
@router.get("/summary")
def get_budget_summary(
month: int = Query(..., ge=1, le=12, description="Month (1-12)"),
year: int = Query(..., ge=2020, le=2030, description="Year"),
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""
Get budget summary for a specific month and year.
Shows budget vs actual spending with category breakdown.
"""
service = BudgetService(db)
return service.get_budget_summary(user_id, month, year)
@router.put("/{budget_id}", response_model=BudgetResponse)
def update_budget(
budget_id: int,
amount: float,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""Update budget amount."""
service = BudgetService(db)
return service.update_budget(budget_id, amount, user_id)
@router.delete("/{budget_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_budget(
budget_id: int,
db: Session = Depends(get_db),
user_id: int = Depends(get_current_user_id)
):
"""Delete a budget."""
service = BudgetService(db)
service.delete_budget(budget_id, user_id)
return None # routers/categories.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from repositories.category import CategoryRepository
from models.category import Category
from schemas.category import CategoryCreate, CategoryResponse
from core.database import get_db
router = APIRouter(prefix="/categories", tags=["categories"])
@router.get("/", response_model=List[CategoryResponse])
def get_categories(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db)
):
"""Get all expense categories."""
repo = CategoryRepository(Category, db)
return repo.get_all(skip, limit)
@router.post("/", response_model=CategoryResponse, status_code=status.HTTP_201_CREATED)
def create_category(
category: CategoryCreate,
db: Session = Depends(get_db)
):
"""Create a new expense category."""
repo = CategoryRepository(Category, db)
# Check if category already exists
existing = repo.get_by_name(category.name)
if existing:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Category with this name already exists"
)
return repo.create(category.dict())
@router.get("/{category_id}", response_model=CategoryResponse)
def get_category(
category_id: int,
db: Session = Depends(get_db)
):
"""Get a specific category by ID."""
repo = CategoryRepository(Category, db)
category = repo.get(category_id)
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found"
)
return category # main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routers import expenses, budgets, categories
from core.database import engine
from models.base import Base
# Create tables
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Expense Management API",
description="A comprehensive expense management system",
version="1.0.0"
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
app.include_router(expenses.router)
app.include_router(budgets.router)
app.include_router(categories.router)
@app.get("/")
def read_root():
return {"message": "Welcome to Expense Management API"}
@app.get("/health")
def health_check():
return {"status": "healthy"}Learning Objectives:
Tasks:
# tests/test_api/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app
from core.database import get_db
from models.base import Base
from models.user import User
from models.category import Category
from models.expense import Expense
from datetime import datetime
# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_api.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def override_get_db():
try:
db = TestingSessionLocal()
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture(scope="function")
def test_db():
Base.metadata.create_all(bind=engine)
yield
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def client():
return TestClient(app)
@pytest.fixture
def test_user(test_db):
db = TestingSessionLocal()
user = User(email="test@example.com", full_name="Test User", department="Engineering")
db.add(user)
db.commit()
db.refresh(user)
db.close()
return user
@pytest.fixture
def test_category(test_db):
db = TestingSessionLocal()
category = Category(name="Food & Dining", description="Restaurant and food expenses")
db.add(category)
db.commit()
db.refresh(category)
db.close()
return category
@pytest.fixture
def test_expense(test_db, test_user, test_category):
db = TestingSessionLocal()
expense = Expense(
amount=25.50,
description="Lunch at restaurant",
expense_date=datetime(2024, 3, 15, 12, 30),
merchant_name="Pizza Palace",
receipt_number="R12345",
user_id=test_user.id,
category_id=test_category.id
)
db.add(expense)
db.commit()
db.refresh(expense)
db.close()
return expense # tests/test_api/test_expenses.py
import pytest
from fastapi.testclient import TestClient
from datetime import datetime
class TestExpenseAPI:
def test_create_expense_success(self, client, test_user, test_category):
"""Test successful expense creation"""
expense_data = {
"amount": 99.99,
"description": "Business lunch",
"expense_date": "2024-03-15T12:00:00",
"merchant_name": "Restaurant ABC",
"receipt_number": "REC001",
"category_id": test_category.id
}
response = client.post("/expenses/", json=expense_data)
assert response.status_code == 201
data = response.json()
assert data["amount"] == 99.99
assert data["description"] == "Business lunch"
assert data["merchant_name"] == "Restaurant ABC"
assert data["category_id"] == test_category.id
assert data["user_id"] == test_user.id
assert "id" in data
assert "created_at" in data
def test_create_expense_negative_amount(self, client, test_category):
"""Test expense creation with negative amount"""
expense_data = {
"amount": -50.0,
"description": "Invalid expense",
"expense_date": "2024-03-15T12:00:00",
"category_id": test_category.id
}
response = client.post("/expenses/", json=expense_data)
assert response.status_code == 422
assert "Amount must be positive" in str(response.json())
def test_create_expense_invalid_category(self, client):
"""Test expense creation with non-existent category"""
expense_data = {
"amount": 99.99,
"description": "Business lunch",
"expense_date": "2024-03-15T12:00:00",
"category_id": 999 # Non-existent category
}
response = client.post("/expenses/", json=expense_data)
assert response.status_code == 404
assert "Category not found" in response.json()["detail"]
def test_get_expenses(self, client, test_expense):
"""Test retrieving user expenses"""
response = client.get("/expenses/")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) >= 1
expense = data[0]
assert expense["id"] == test_expense.id
assert expense["amount"] == test_expense.amount
def test_get_expenses_with_pagination(self, client, test_user, test_category):
"""Test expense pagination"""
# Create multiple expenses
for i in range(5):
expense_data = {
"amount": 10.0 + i,
"description": f"Expense {i}",
"expense_date": "2024-03-15T12:00:00",
"category_id": test_category.id
}
client.post("/expenses/", json=expense_data)
# Test pagination
response = client.get("/expenses/?skip=0&limit=3")
assert response.status_code == 200
data = response.json()
assert len(data) <= 3
def test_get_expenses_with_date_filter(self, client, test_user, test_category):
"""Test expense filtering by date range"""
# Create expenses with different dates
expenses_data = [
{
"amount": 50.0,
"description": "March expense",
"expense_date": "2024-03-15T12:00:00",
"category_id": test_category.id
},
{
"amount": 75.0,
"description": "April expense",
"expense_date": "2024-04-15T12:00:00",
"category_id": test_category.id
}
]
for expense_data in expenses_data:
client.post("/expenses/", json=expense_data)
# Filter for March only
response = client.get("/expenses/?start_date=2024-03-01&end_date=2024-03-31")
assert response.status_code == 200
data = response.json()
# Should only return March expenses
for expense in data:
expense_date = datetime.fromisoformat(expense["expense_date"].replace("Z", "+00:00"))
assert expense_date.month == 3
def test_get_expense_by_id(self, client, test_expense):
"""Test retrieving specific expense"""
response = client.get(f"/expenses/{test_expense.id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == test_expense.id
assert data["amount"] == test_expense.amount
def test_get_expense_not_found(self, client):
"""Test retrieving non-existent expense"""
response = client.get("/expenses/999")
assert response.status_code == 404
assert "Expense not found" in response.json()["detail"]
def test_update_expense(self, client, test_expense):
"""Test updating an expense"""
update_data = {
"amount": 150.0,
"description": "Updated description"
}
response = client.put(f"/expenses/{test_expense.id}", json=update_data)
assert response.status_code == 200
data = response.json()
assert data["amount"] == 150.0
assert data["description"] == "Updated description"
def test_delete_expense(self, client, test_expense):
"""Test deleting an expense"""
response = client.delete(f"/expenses/{test_expense.id}")
assert response.status_code == 204
# Verify expense is deleted
get_response = client.get(f"/expenses/{test_expense.id}")
assert get_response.status_code == 404
def test_get_monthly_summary(self, client, test_user, test_category):
"""Test monthly summary endpoint"""
# Create test expenses for March 2024
expenses_data = [
{
"amount": 100.0,
"description": "Expense 1",
"expense_date": "2024-03-05T12:00:00",
"category_id": test_category.id
},
{
"amount": 200.0,
"description": "Expense 2",
"expense_date": "2024-03-15T12:00:00",
"category_id": test_category.id
}
]
for expense_data in expenses_data:
client.post("/expenses/", json=expense_data)
response = client.get("/expenses/summary/monthly?month=3&year=2024")
assert response.status_code == 200
data = response.json()
assert data["month"] == 3
assert data["year"] == 2024
assert data["total_amount"] == 300.0
assert data["total_expenses"] == 2
assert test_category.id in data["category_breakdown"] # tests/test_api/test_budgets.py
import pytest
from fastapi.testclient import TestClient
class TestBudgetAPI:
def test_create_budget_success(self, client, test_user, test_category):
"""Test successful budget creation"""
budget_data = {
"amount": 500.0,
"month": 3,
"year": 2024,
"category_id": test_category.id
}
response = client.post("/budgets/", json=budget_data)
assert response.status_code == 201
data = response.json()
assert data["amount"] == 500.0
assert data["month"] == 3
assert data["year"] == 2024
assert data["category_id"] == test_category.id
assert data["user_id"] == test_user.id
def test_create_budget_negative_amount(self, client, test_category):
"""Test budget creation with negative amount"""
budget_data = {
"amount": -100.0,
"month": 3,
"year": 2024,
"category_id": test_category.id
}
response = client.post("/budgets/", json=budget_data)
assert response.status_code == 422
assert "Budget amount must be positive" in str(response.json())
def test_create_budget_invalid_month(self, client, test_category):
"""Test budget creation with invalid month"""
budget_data = {
"amount": 500.0,
"month": 13, # Invalid month
"year": 2024,
"category_id": test_category.id
}
response = client.post("/budgets/", json=budget_data)
assert response.status_code == 422
assert "Month must be between 1 and 12" in str(response.json())
def test_create_duplicate_budget(self, client, test_user, test_category):
"""Test creating duplicate budget for same period"""
budget_data = {
"amount": 500.0,
"month": 3,
"year": 2024,
"category_id": test_category.id
}
# Create first budget
response1 = client.post("/budgets/", json=budget_data)
assert response1.status_code == 201
# Try to create duplicate
response2 = client.post("/budgets/", json=budget_data)
assert response2.status_code == 400
assert "already exists" in response2.json()["detail"]
def test_get_budget_summary(self, client, test_user, test_category):
"""Test budget summary endpoint"""
# Create budget
budget_data = {
"amount": 500.0,
"month": 3,
"year": 2024,
"category_id": test_category.id
}
client.post("/budgets/", json=budget_data)
# Create some expenses
expense_data = {
"amount": 200.0,
"description": "Test expense",
"expense_date": "2024-03-15T12:00:00",
"category_id": test_category.id
}
client.post("/expenses/", json=expense_data)
response = client.get("/budgets/summary?month=3&year=2024")
assert response.status_code == 200
data = response.json()
assert data["month"] == 3
assert data["year"] == 2024
assert data["total_budget"] == 500.0
assert data["total_spent"] == 200.0
assert data["remaining"] == 300.0
assert data["is_over_budget"] is False
# Check category breakdown
assert len(data["categories"]) == 1
category_data = data["categories"][0]
assert category_data["budget_amount"] == 500.0
assert category_data["spent_amount"] == 200.0
assert category_data["percentage_used"] == 40.0 # tests/test_api/test_categories.py
import pytest
from fastapi.testclient import TestClient
class TestCategoryAPI:
def test_create_category_success(self, client):
"""Test successful category creation"""
category_data = {
"name": "Transportation",
"description": "Car, bus, train expenses"
}
response = client.post("/categories/", json=category_data)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Transportation"
assert data["description"] == "Car, bus, train expenses"
assert "id" in data
def test_create_duplicate_category(self, client, test_category):
"""Test creating category with duplicate name"""
category_data = {
"name": test_category.name,
"description": "Duplicate category"
}
response = client.post("/categories/", json=category_data)
assert response.status_code == 400
assert "already exists" in response.json()["detail"]
def test_get_categories(self, client, test_category):
"""Test retrieving all categories"""
response = client.get("/categories/")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) >= 1
# Find our test category
found_category = next((cat for cat in data if cat["id"] == test_category.id), None)
assert found_category is not None
assert found_category["name"] == test_category.name
def test_get_category_by_id(self, client, test_category):
"""Test retrieving specific category"""
response = client.get(f"/categories/{test_category.id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == test_category.id
assert data["name"] == test_category.name
def test_get_category_not_found(self, client):
"""Test retrieving non-existent category"""
response = client.get("/categories/999")
assert response.status_code == 404
assert "Category not found" in response.json()["detail"] # tests/test_api/test_integration.py
import pytest
from fastapi.testclient import TestClient
class TestExpenseWorkflow:
def test_complete_expense_workflow(self, client):
"""Test complete expense management workflow"""
# 1. Create a category
category_data = {
"name": "Business Meals",
"description": "Client dinners and business lunches"
}
category_response = client.post("/categories/", json=category_data)
assert category_response.status_code == 201
category_id = category_response.json()["id"]
# 2. Create a budget for the category
budget_data = {
"amount": 1000.0,
"month": 3,
"year": 2024,
"category_id": category_id
}
budget_response = client.post("/budgets/", json=budget_data)
assert budget_response.status_code == 201
# 3. Create expenses within budget
expense1_data = {
"amount": 300.0,
"description": "Client dinner",
"expense_date": "2024-03-10T19:00:00",
"merchant_name": "Fine Dining Restaurant",
"receipt_number": "R001",
"category_id": category_id
}
expense1_response = client.post("/expenses/", json=expense1_data)
assert expense1_response.status_code == 201
expense1_id = expense1_response.json()["id"]
expense2_data = {
"amount": 150.0,
"description": "Business lunch",
"expense_date": "2024-03-15T12:00:00",
"merchant_name": "Bistro Downtown",
"receipt_number": "R002",
"category_id": category_id
}
expense2_response = client.post("/expenses/", json=expense2_data)
assert expense2_response.status_code == 201
# 4. Check budget summary
summary_response = client.get("/budgets/summary?month=3&year=2024")
assert summary_response.status_code == 200
summary = summary_response.json()
assert summary["total_budget"] == 1000.0
assert summary["total_spent"] == 450.0
assert summary["remaining"] == 550.0
# 5. Try to create expense that exceeds budget
large_expense_data = {
"amount": 600.0, # This would exceed remaining budget
"description": "Expensive dinner",
"expense_date": "2024-03-20T19:00:00",
"category_id": category_id
}
large_expense_response = client.post("/expenses/", json=large_expense_data)
assert large_expense_response.status_code == 400
assert "exceeds monthly budget limit" in large_expense_response.json()["detail"]
# 6. Update an expense
update_data = {
"amount": 250.0,
"description": "Updated client dinner"
}
update_response = client.put(f"/expenses/{expense1_id}", json=update_data)
assert update_response.status_code == 200
# 7. Get monthly summary
monthly_response = client.get("/expenses/summary/monthly?month=3&year=2024")
assert monthly_response.status_code == 200
monthly_summary = monthly_response.json()
assert monthly_summary["total_expenses"] == 2
assert category_id in monthly_summary["category_breakdown"]
# 8. Delete an expense
delete_response = client.delete(f"/expenses/{expense1_id}")
assert delete_response.status_code == 204
# 9. Verify expense is deleted
get_deleted_response = client.get(f"/expenses/{expense1_id}")
assert get_deleted_response.status_code == 404Deliverable: Complete API layer with comprehensive testing
Learning Objectives:
Tasks:
# core/exceptions.py
from fastapi import HTTPException, status
class ExpenseManagementException(Exception):
"""Base exception for expense management system"""
pass
class BudgetExceededException(ExpenseManagementException):
"""Raised when expense exceeds budget limit"""
def __init__(self, remaining_budget: float):
self.remaining_budget = remaining_budget
super().__init__(f"Expense exceeds budget. Remaining: ${remaining_budget:.2f}")
class DuplicateResourceException(ExpenseManagementException):
"""Raised when trying to create duplicate resource"""
pass
class UnauthorizedAccessException(ExpenseManagementException):
"""Raised when user tries to access unauthorized resource"""
pass
# main.py - Add exception handlers
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from core.exceptions import BudgetExceededException, DuplicateResourceException
@app.exception_handler(BudgetExceededException)
async def budget_exceeded_handler(request: Request, exc: BudgetExceededException):
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={
"detail": str(exc),
"error_type": "budget_exceeded",
"remaining_budget": exc.remaining_budget
}
)
@app.exception_handler(DuplicateResourceException)
async def duplicate_resource_handler(request: Request, exc: DuplicateResourceException):
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={
"detail": str(exc),
"error_type": "duplicate_resource"
}
)Learning Objectives:
Tasks:
# Update main.py with better documentation
app = FastAPI(
title="Expense Management API",
description="""
A comprehensive personal expense management system that helps users:
* **Track expenses** with detailed categorization
* **Set and monitor budgets** for different expense categories
* **Generate reports** and summaries for financial planning
* **Manage categories** for better expense organization
## Features
* Create, read, update, and delete expenses
* Set monthly budgets by category
* Real-time budget tracking and alerts
* Monthly and yearly expense summaries
* Category-wise expense breakdown
## Authentication
This API uses simplified user identification for training purposes.
In production, implement proper JWT-based authentication.
""",
version="1.0.0",
contact={
"name": "Expense Management Team",
"email": "support@expensemanager.com",
},
license_info={
"name": "MIT",
},
) # middleware/performance.py
import time
from fastapi import Request
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return responsefastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.0
pytest==7.4.3
pytest-asyncio==0.21.1
python-multipart==0.0.6This training plan provides a structured approach to learning FastAPI with repository pattern through systematic development and testing at each layer. The hands-on approach ensures practical understanding of enterprise-level development patterns.