#!/usr/bin/env python3
"""Expense Tracker API - Main Application"""
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Optional
from datetime import datetime
import uvicorn
from models import (
Transaction, TransactionCreate, TransactionUpdate,
TransactionType, Category, TransactionSummary, CategorySummary
)
from store import TransactionStore
from config import settings
# Create FastAPI application
app = FastAPI(
title=settings.API_TITLE,
version=settings.API_VERSION,
docs_url="/docs",
redoc_url="/redoc"
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOW_ORIGINS,
allow_credentials=settings.ALLOW_CREDENTIALS,
allow_methods=settings.ALLOW_METHODS,
allow_headers=settings.ALLOW_HEADERS,
)
# Initialize transaction store
transaction_store = TransactionStore()
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"service": "expense-tracker-api",
"version": settings.API_VERSION,
"timestamp": datetime.now().isoformat(),
"transactions_count": len(transaction_store.get_all())
}
@app.get("/transactions", response_model=List[Transaction])
async def get_transactions(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=500),
type: Optional[TransactionType] = None,
category: Optional[Category] = None
):
"""Get all transactions with pagination and filtering"""
transactions = transaction_store.get_all()
# Apply filters
if type:
transactions = [t for t in transactions if t.type == type]
if category:
transactions = [t for t in transactions if t.category == category]
return transactions[skip:skip + limit]
@app.get("/transactions/search", response_model=List[Transaction])
async def search_transactions(q: str = Query(..., min_length=1)):
"""Search transactions by title, description, category, or tags"""
return transaction_store.search(q)
@app.get("/transactions/{transaction_id}", response_model=Transaction)
async def get_transaction(transaction_id: str):
"""Get a specific transaction by ID"""
transaction = transaction_store.get_by_id(transaction_id)
if not transaction:
raise HTTPException(status_code=404, detail="Transaction not found")
return transaction
@app.post("/transactions", response_model=Transaction, status_code=201)
async def create_transaction(transaction_data: TransactionCreate):
"""Create a new transaction"""
return transaction_store.create(transaction_data)
@app.put("/transactions/{transaction_id}", response_model=Transaction)
async def update_transaction(transaction_id: str, transaction_data: TransactionUpdate):
"""Update an existing transaction"""
if not transaction_store.get_by_id(transaction_id):
raise HTTPException(status_code=404, detail="Transaction not found")
return transaction_store.update(transaction_id, transaction_data)
@app.delete("/transactions/{transaction_id}")
async def delete_transaction(transaction_id: str):
"""Delete a transaction"""
if not transaction_store.delete(transaction_id):
raise HTTPException(status_code=404, detail="Transaction not found")
return {"message": "Transaction deleted successfully", "transaction_id": transaction_id}
@app.get("/summary", response_model=TransactionSummary)
async def get_summary():
"""Get transaction summary statistics"""
summary_data = transaction_store.get_summary()
return TransactionSummary(**summary_data)
@app.get("/summary/categories", response_model=List[CategorySummary])
async def get_category_summary():
"""Get summary by category"""
category_data = transaction_store.get_category_summary()
return [CategorySummary(**item) for item in category_data]
@app.get("/transactions/type/{transaction_type}", response_model=List[Transaction])
async def get_by_type(transaction_type: TransactionType):
"""Get transactions by type (income or expense)"""
return transaction_store.get_by_type(transaction_type)
@app.get("/transactions/category/{category}", response_model=List[Transaction])
async def get_by_category(category: Category):
"""Get transactions by category"""
return transaction_store.get_by_category(category)
@app.get("/transactions/date-range")
async def get_by_date_range(
start_date: datetime = Query(...),
end_date: datetime = Query(...)
):
"""Get transactions within a date range"""
if start_date > end_date:
raise HTTPException(status_code=400, detail="Start date must be before end date")
return transaction_store.get_by_date_range(start_date, end_date)
if __name__ == "__main__":
print(f"🚀 Starting Expense Tracker API...")
print(f"🌐 Server: http://{settings.HOST}:{settings.PORT}")
print(f"📚 API Docs: http://{settings.HOST}:{settings.PORT}/docs")
uvicorn.run(
"main:app",
host=settings.HOST,
port=settings.PORT,
reload=True
)
