11. Data Models & Database Schema

11.1 Database Technology

Advantages

• ? Zero configuration
• ? No server required
• ? Fully offline
• ? File-based (easy backup)
• ? ACID compliant
• ? Cross-platform
• ? Fast for local operations

11.2 Complete Database Schema

11.2.1 Calculations Table

CREATE TABLE calculations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    amount DECIMAL(20, 2) NOT NULL,
    currency VARCHAR(3) NOT NULL,
    mode VARCHAR(20) NOT NULL,
    breakdown TEXT NOT NULL,  -- JSON string
    summary TEXT NOT NULL,     -- JSON string
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    
    -- Indexes for performance
    INDEX idx_currency (currency),
    INDEX idx_timestamp (timestamp DESC),
    INDEX idx_amount (amount)
);

SQLAlchemy Model

from sqlalchemy import Column, Integer, String, DateTime, Text, Numeric, Index
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, autoincrement=True)
    amount = Column(Numeric(precision=20, scale=2), nullable=False, index=True)
    currency = Column(String(3), nullable=False, index=True)
    mode = Column(String(20), nullable=False)
    breakdown = Column(Text, nullable=False)  # JSON: List[{denomination, type, count, total_value}]
    summary = Column(Text, nullable=False)     # JSON: {total_notes, total_coins, total_denominations}
    timestamp = Column(DateTime, default=datetime.utcnow, nullable=False, index=True)
    
    # Composite indexes
    __table_args__ = (
        Index('idx_currency_timestamp', 'currency', 'timestamp'),
    )
    
    def __repr__(self):
        return f""

Column Details

11.2.2 Settings Table

CREATE TABLE settings (
    id INTEGER PRIMARY KEY DEFAULT 1,  -- Single row table
    theme VARCHAR(10) DEFAULT 'light' NOT NULL,
    language VARCHAR(2) DEFAULT 'en' NOT NULL,
    default_currency VARCHAR(3) DEFAULT 'INR' NOT NULL,
    default_mode VARCHAR(20) DEFAULT 'greedy' NOT NULL,
    auto_save_history BOOLEAN DEFAULT 1 NOT NULL,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
    
    -- Ensure only one settings row
    CHECK (id = 1)
);

SQLAlchemy Model

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

Column Details

11.3 Sample Data & Queries

Common Queries

# Get last 10 calculations
recent = db.query(Calculation)\
    .order_by(Calculation.timestamp.desc())\
    .limit(10)\
    .all()

# Get all INR calculations
inr_calcs = db.query(Calculation)\
    .filter(Calculation.currency == 'INR')\
    .order_by(Calculation.timestamp.desc())\
    .all()

def get_paginated_history(page: int = 1, per_page: int = 50):
    offset = (page - 1) * per_page
    
    results = db.query(Calculation)\
        .order_by(Calculation.timestamp.desc())\
        .offset(offset)\
        .limit(per_page)\
        .all()
    
    total = db.query(Calculation).count()
    
    return {
        'items': results,
        'total': total,
        'page': page,
        'pages': (total + per_page - 1) // per_page
    }

from sqlalchemy import func

# Total calculations by currency
stats = db.query(
    Calculation.currency,
    func.count(Calculation.id).label('count'),
    func.sum(Calculation.amount).label('total_amount')
)\
    .group_by(Calculation.currency)\
    .all()

# Output:
# [('INR', 150, Decimal('450000.00')),
#  ('USD', 50, Decimal('125000.00')),
#  ...]

from datetime import datetime, timedelta

# Last 7 days
start_date = datetime.utcnow() - timedelta(days=7)

recent_week = db.query(Calculation)\
    .filter(Calculation.timestamp >= start_date)\
    .order_by(Calculation.timestamp.desc())\
    .all()

# Delete by ID
calc = db.query(Calculation).filter(Calculation.id == 42).first()
if calc:
    db.delete(calc)
    db.commit()

# Delete all calculations
db.query(Calculation).delete()
db.commit()

# Always returns single row
settings = db.query(Settings).first()

if not settings:
    # Initialize with defaults
    settings = Settings(
        id=1,
        theme="light",
        language="en",
        default_currency="INR",
        default_mode="greedy",
        auto_save_history=True
    )
    db.add(settings)
    db.commit()

settings = db.query(Settings).first()
settings.theme = "dark"
settings.language = "hi"
db.commit()

11.4 Database Initialization

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.db.models import Base, Settings

# Create engine
DATABASE_URL = "sqlite:///./currency_calculator.db"
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLite specific
    echo=False  # Set True for SQL logging
)

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

# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Dependency for FastAPI
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Initialize default settings
def init_database():
    """Initialize database with default settings."""
    db = SessionLocal()
    
    try:
        # Check if settings exist
        if not db.query(Settings).first():
            default_settings = Settings(
                id=1,
                theme="light",
                language="en",
                default_currency="INR",
                default_mode="greedy",
                auto_save_history=True
            )
            db.add(default_settings)
            db.commit()
            print("? Default settings initialized")
    finally:
        db.close()

# Call on startup
init_database()

11.5 JSON Field Structures

Breakdown JSON Format

[
  {
    "denomination": 500,
    "type": "note",
    "count": 3,
    "total_value": 1500
  },
  {
    "denomination": 200,
    "type": "note",
    "count": 1,
    "total_value": 200
  },
  {
    "denomination": 50,
    "type": "note",
    "count": 1,
    "total_value": 50
  },
  {
    "denomination": 5,
    "type": "coin",
    "count": 2,
    "total_value": 10
  }
]

Summary JSON Format

{
  "total_notes": 5,
  "total_coins": 2,
  "total_denominations": 4
}

11.6 Backup & Restore

Manual Backup

# Simple file copy
Copy-Item currency_calculator.db -Destination backup_$(Get-Date -Format 'yyyyMMdd_HHmmss').db

Automated Backup Script

import shutil
from datetime import datetime
from pathlib import Path

def backup_database():
    """Create timestamped database backup."""
    db_path = Path("currency_calculator.db")
    
    if not db_path.exists():
        print("Database not found")
        return
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_path = Path(f"backups/currency_calculator_{timestamp}.db")
    backup_path.parent.mkdir(exist_ok=True)
    
    shutil.copy2(db_path, backup_path)
    print(f"? Backup created: {backup_path}")

# Schedule daily backups
if __name__ == "__main__":
    backup_database()

Export to CSV

import csv
from app.db.database import SessionLocal
from app.db.models import Calculation

def export_to_csv(output_file: str):
    """Export all calculations to CSV."""
    db = SessionLocal()
    
    try:
        calcs = db.query(Calculation).order_by(Calculation.timestamp.desc()).all()
        
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            
            # Header
            writer.writerow(['ID', 'Amount', 'Currency', 'Mode', 'Timestamp', 'Breakdown', 'Summary'])
            
            # Data
            for calc in calcs:
                writer.writerow([
                    calc.id,
                    calc.amount,
                    calc.currency,
                    calc.mode,
                    calc.timestamp,
                    calc.breakdown,
                    calc.summary
                ])
        
        print(f"? Exported {len(calcs)} calculations to {output_file}")
    finally:
        db.close()