6. Backend Functional Logic

6.1 FastAPI Application Structure

File: packages/local-backend/app/main.py

Application Initialization

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api import calculations, bulk, ocr, history, settings
from app.db.database import engine, Base

# Create tables
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Currency Denomination Calculator API",
    version="1.0.0",
    description="Local backend for denomination calculations"
)

# CORS configuration (Electron app)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "app://"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
app.include_router(calculations.router, prefix="/api/v1", tags=["calculations"])
app.include_router(bulk.router, prefix="/api/v1", tags=["bulk"])
app.include_router(ocr.router, prefix="/api/v1", tags=["ocr"])
app.include_router(history.router, prefix="/api/v1", tags=["history"])
app.include_router(settings.router, prefix="/api/v1", tags=["settings"])

6.2 Complete API Endpoints

6.2.1 Calculations Endpoints

File: packages/local-backend/app/api/calculations.py (530+ lines)

POST /api/v1/calculate

Purpose: Calculate denomination breakdown for a single amount

class CalculationRequest(BaseModel):
    amount: Decimal = Field(..., gt=0, description="Amount to calculate")
    currency: str = Field(..., pattern="^(INR|USD|EUR|GBP)$")
    mode: str = Field(
        default="greedy",
        pattern="^(greedy|balanced|minimize_large|minimize_small)$"
    )
    save_to_history: bool = Field(default=True)

class CalculationResponse(BaseModel):
    calculation_id: int
    amount: Decimal
    currency: str
    mode: str
    breakdown: List[DenominationItem]
    summary: Summary
    timestamp: datetime

class DenominationItem(BaseModel):
    denomination: Decimal
    type: str  # "note" or "coin"
    count: int
    total_value: Decimal

class Summary(BaseModel):
    total_notes: int
    total_coins: int
    total_denominations: int

Business Logic Implementation

@router.post("/calculate", response_model=CalculationResponse)
async def calculate(
    request: CalculationRequest,
    db: Session = Depends(get_db)
):
    # 1. Validate amount (additional business rules)
    if request.amount > Decimal("999999999999999"):
        raise HTTPException(400, "Amount too large")
    
    # 2. Initialize engine
    engine = DenominationEngine()
    
    # 3. Calculate breakdown
    result = engine.calculate(
        amount=request.amount,
        currency=request.currency,
        mode=request.mode
    )
    
    # 4. Save to history if requested
    if request.save_to_history:
        calculation = Calculation(
            amount=request.amount,
            currency=request.currency,
            mode=request.mode,
            breakdown=json.dumps(result['breakdown']),
            summary=json.dumps(result['summary']),
            timestamp=datetime.utcnow()
        )
        db.add(calculation)
        db.commit()
        db.refresh(calculation)
        calculation_id = calculation.id
    else:
        calculation_id = -1  # Not saved
    
    # 5. Return response
    return CalculationResponse(
        calculation_id=calculation_id,
        amount=request.amount,
        currency=request.currency,
        mode=request.mode,
        breakdown=result['breakdown'],
        summary=result['summary'],
        timestamp=datetime.utcnow()
    )

Error Responses

POST /api/v1/calculate-batch

Purpose: Calculate multiple amounts in one request

class BatchCalculationRequest(BaseModel):
    items: List[CalculationRequest] = Field(..., max_items=1000)

class BatchCalculationResponse(BaseModel):
    results: List[CalculationResponse]
    summary: BatchSummary

class BatchSummary(BaseModel):
    total_items: int
    successful: int
    failed: int
    total_time_seconds: float

6.2.2 Bulk Upload Endpoints

File: packages/local-backend/app/api/bulk.py

POST /api/v1/bulk/upload

Purpose: Process bulk upload files (CSV, PDF, Word, Images)

File Validation

# Size limits
MAX_FILE_SIZE = {
    'csv': 10 * 1024 * 1024,      # 10 MB
    'pdf': 50 * 1024 * 1024,      # 50 MB
    'docx': 10 * 1024 * 1024,     # 10 MB
    'image': 50 * 1024 * 1024,    # 50 MB
}

# Extension validation
ALLOWED_EXTENSIONS = {
    'csv': ['.csv'],
    'pdf': ['.pdf'],
    'word': ['.docx', '.doc'],
    'image': ['.jpg', '.jpeg', '.png', '.tiff', '.bmp']
}

def validate_file(file: UploadFile) -> str:
    # Check extension
    ext = Path(file.filename).suffix.lower()
    file_type = None
    
    for type_name, exts in ALLOWED_EXTENSIONS.items():
        if ext in exts:
            file_type = type_name
            break
    
    if not file_type:
        raise HTTPException(400, f"Unsupported file type: {ext}")
    
    # Check size
    file.file.seek(0, 2)  # Seek to end
    size = file.file.tell()
    file.file.seek(0)  # Reset
    
    if size > MAX_FILE_SIZE[file_type]:
        raise HTTPException(400, f"File too large")
    
    if size == 0:
        raise HTTPException(400, "File is empty")
    
    return file_type

Processing Logic

@router.post("/bulk/upload", response_model=BulkUploadResponse)
async def bulk_upload(
    file: UploadFile = File(...),
    save_to_history: bool = Form(default=True),
    db: Session = Depends(get_db)
):
    # 1. Validate file
    file_type = validate_file(file)
    
    # 2. Save temporary file
    temp_path = Path(f"/tmp/{file.filename}")
    with temp_path.open("wb") as f:
        shutil.copyfileobj(file.file, f)
    
    try:
        # 3. Route to appropriate processor
        if file_type == 'csv':
            data = process_csv(temp_path)
        elif file_type == 'pdf':
            data = await process_pdf_ocr(temp_path)
        elif file_type == 'word':
            data = process_word(temp_path)
        elif file_type == 'image':
            data = await process_image_ocr(temp_path)
        
        # 4. Perform calculations
        results = []
        for row in data:
            try:
                calc_result = await calculate(
                    CalculationRequest(**row),
                    db
                )
                results.append({
                    'status': 'success',
                    'row_number': row['row_number'],
                    'result': calc_result
                })
            except Exception as e:
                results.append({
                    'status': 'failed',
                    'row_number': row['row_number'],
                    'error': str(e)
                })
        
        # 5. Compute summary
        successful = sum(1 for r in results if r['status'] == 'success')
        
        return BulkUploadResponse(
            filename=file.filename,
            total_rows=len(results),
            successful_rows=successful,
            failed_rows=len(results) - successful,
            results=results
        )
    
    finally:
        # 6. Cleanup
        temp_path.unlink(missing_ok=True)

6.2.3 History Endpoints

GET /api/v1/history

Purpose: Get paginated calculation history

Business Logic

@router.get("/history", response_model=HistoryResponse)
async def get_history(
    page: int = 1,
    per_page: int = 50,
    currency: Optional[str] = None,
    sort_by: str = "timestamp",
    sort_order: str = "desc",
    db: Session = Depends(get_db)
):
    # 1. Build query
    query = db.query(Calculation)
    
    # 2. Apply filters
    if currency:
        query = query.filter(Calculation.currency == currency)
    
    # 3. Apply sorting
    if sort_by == "timestamp":
        order_col = Calculation.timestamp
    elif sort_by == "amount":
        order_col = Calculation.amount
    
    if sort_order == "desc":
        query = query.order_by(order_col.desc())
    else:
        query = query.order_by(order_col.asc())
    
    # 4. Count total
    total_items = query.count()
    
    # 5. Paginate
    offset = (page - 1) * per_page
    items = query.offset(offset).limit(per_page).all()
    
    # 6. Format response
    total_pages = (total_items + per_page - 1) // per_page
    
    return HistoryResponse(
        items=items,
        pagination=Pagination(
            page=page,
            per_page=per_page,
            total_items=total_items,
            total_pages=total_pages,
            has_next=page < total_pages,
            has_prev=page > 1
        )
    )

DELETE /api/v1/history/{id}

Purpose: Delete a specific calculation

DELETE /api/v1/history/clear

Purpose: Delete all calculations (requires confirmation)

6.2.4 Settings Endpoints

GET /api/v1/settings

Response Schema:

class SettingsResponse(BaseModel):
    theme: str  # "light", "dark", "system"
    language: str  # "en", "hi", "es", "fr", "de"
    default_currency: str  # "INR", "USD", "EUR", "GBP"
    default_mode: str  # "greedy", "balanced", etc.
    auto_save_history: bool

PUT /api/v1/settings

Purpose: Update user settings

POST /api/v1/settings/reset

Purpose: Reset all settings to defaults

6.3 Error Handling Strategy

Global Exception Handler

from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "detail": exc.errors(),
            "body": exc.body
        }
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "detail": exc.detail
        }
    )

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "detail": "Internal server error"
        }
    )

Custom Exceptions

class CalculationEngineError(Exception):
    """Raised when denomination engine fails"""
    pass

class OCRProcessingError(Exception):
    """Raised when OCR processing fails"""
    pass

class FileValidationError(Exception):
    """Raised when file validation fails"""
    pass

6.4 Logging Configuration

File: packages/local-backend/app/utils/logger.py

import logging
from logging.handlers import RotatingFileHandler

def setup_logger():
    """Configure application-wide logging."""
    logger = logging.getLogger("currency_calculator")
    logger.setLevel(logging.INFO)
    
    # Console handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_format = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    console_handler.setFormatter(console_format)
    
    # File handler (rotating)
    file_handler = RotatingFileHandler(
        'logs/app.log',
        maxBytes=10 * 1024 * 1024,  # 10 MB
        backupCount=5
    )
    file_handler.setLevel(logging.DEBUG)
    file_format = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
    )
    file_handler.setFormatter(file_format)
    
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    
    return logger

logger = setup_logger()

Usage in Endpoints

from app.utils.logger import logger

@router.post("/calculate")
async def calculate(request: CalculationRequest, db: Session = Depends(get_db)):
    logger.info(f"Calculation request: amount={request.amount}, currency={request.currency}")
    
    try:
        result = engine.calculate(...)
        logger.info(f"Calculation successful: id={calc_id}")
        return result
    except Exception as e:
        logger.error(f"Calculation failed: {e}", exc_info=True)
        raise HTTPException(500, "Calculation failed")

6.5 Database Models

File: packages/local-backend/app/db/models.py

from sqlalchemy import Column, Integer, String, DateTime, Text, Boolean, Numeric
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Calculation(Base):
    __tablename__ = "calculations"
    
    id = Column(Integer, primary_key=True, index=True)
    amount = Column(Numeric(precision=20, scale=2), nullable=False)
    currency = Column(String(3), nullable=False, index=True)
    mode = Column(String(20), nullable=False)
    breakdown = Column(Text, nullable=False)  # JSON string
    summary = Column(Text, nullable=False)  # JSON string
    timestamp = Column(DateTime, default=datetime.utcnow, index=True)

class Settings(Base):
    __tablename__ = "settings"
    
    id = Column(Integer, primary_key=True)
    theme = Column(String(10), default="light")
    language = Column(String(2), default="en")
    default_currency = Column(String(3), default="INR")
    default_mode = Column(String(20), default="greedy")
    auto_save_history = Column(Boolean, default=True)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

Database Connection

# app/db/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./currency_calculator.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite specific
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

6.6 Validation Business Rules

Amount Validation

def validate_amount(amount: Decimal) -> None:
    """Validate amount follows business rules."""
    if amount <= 0:
        raise ValueError("Amount must be greater than 0")
    
    if amount > Decimal("999999999999999"):  # 15 digits
        raise ValueError("Amount too large (max 15 digits)")
    
    # Check decimal places
    if amount.as_tuple().exponent < -2:
        raise ValueError("Amount can have maximum 2 decimal places")

Currency Validation

SUPPORTED_CURRENCIES = {"INR", "USD", "EUR", "GBP"}

def validate_currency(currency: str) -> None:
    """Validate currency is supported."""
    if currency not in SUPPORTED_CURRENCIES:
        raise ValueError(f"Unsupported currency: {currency}")

Mode Validation

SUPPORTED_MODES = {"greedy", "balanced", "minimize_large", "minimize_small"}

def validate_mode(mode: str) -> None:
    """Validate optimization mode is supported."""
    if mode not in SUPPORTED_MODES:
        raise ValueError(f"Unsupported mode: {mode}")

6.7 Data Transformation Logic

Breakdown to Summary

def calculate_summary(breakdown: List[Dict]) -> Dict:
    """Calculate summary statistics from breakdown."""
    total_notes = sum(
        item['count'] for item in breakdown
        if item['type'] == 'note'
    )
    
    total_coins = sum(
        item['count'] for item in breakdown
        if item['type'] == 'coin'
    )
    
    total_denominations = len([
        item for item in breakdown
        if item['count'] > 0
    ])
    
    return {
        'total_notes': total_notes,
        'total_coins': total_coins,
        'total_denominations': total_denominations
    }

JSON Serialization

import json
from decimal import Decimal

class DecimalEncoder(json.JSONEncoder):
    """Custom JSON encoder for Decimal types."""
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return super().default(obj)

# Usage
json_string = json.dumps(breakdown, cls=DecimalEncoder)