Databases & APIs Capstone Projects

# =========================================================================
# PROJECT 1: SQLITE E-COMMERCE INVENTORY & ORDERS ENGINE
# =========================================================================
import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
cur = conn.cursor()

# 1. Enable Foreign Key Constraints in SQLite:
cur.execute("PRAGMA foreign_keys = ON")

# 2. Build Relational Schema:
cur.execute("""
CREATE TABLE products (
    product_id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    price REAL NOT NULL,
    stock INTEGER NOT NULL
)
""")

cur.execute("""
CREATE TABLE orders (
    order_id INTEGER PRIMARY KEY AUTOINCREMENT,
    customer_name TEXT NOT NULL,
    product_id INTEGER NOT NULL,
    quantity INTEGER NOT NULL,
    total_amount REAL NOT NULL,
    order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (product_id) REFERENCES products (product_id)
)
""")

# Seed initial catalog:
cur.executemany("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)", [
    ("MacBook Pro M3", 169999.00, 5),
    ("Mechanical Keyboard", 2499.00, 20),
    ("Ultra-Wide 4K Monitor", 34999.00, 3)
])
conn.commit()

def process_order(customer_name, product_id, quantity):
    """Atomic order checkout transaction with inventory deduction."""
    print(f"\n--- 🛍️ Processing Order: '{customer_name}' buying {quantity}x Product #{product_id} ---")
    try:
        # Check stock:
        cur.execute("SELECT name, price, stock FROM products WHERE product_id = ?", (product_id,))
        product = cur.fetchone()
        if not product:
            raise ValueError("Product does not exist!")
        if product["stock"] < quantity:
            raise ValueError(f"Out of stock! Only {product['stock']} units available.")

        total_cost = product["price"] * quantity

        # Deduct inventory:
        cur.execute("UPDATE products SET stock = stock - ? WHERE product_id = ?", (quantity, product_id))
        
        # Insert Order Record:
        cur.execute("""
        INSERT INTO orders (customer_name, product_id, quantity, total_amount)
        VALUES (?, ?, ?, ?)
        """, (customer_name, product_id, quantity, total_cost))

        conn.commit()
        print(f"✅ Order #{cur.lastrowid} Confirmed! Total: ₹{total_cost:,.2f} ({product['name']})")
    except Exception as err:
        conn.rollback()
        print(f"❌ Order Failed: {err}")

# Test Orders:
process_order("Balaji", 1, 2)  # Success: Buys 2 MacBooks
process_order("Alex", 1, 10)  # Fails: Exceeds remaining stock!
process_order("Chloe", 2, 3)  # Success: Buys 3 Keyboards

# View Final Inventory:
print("\n--- Current Warehouse Stock ---")
cur.execute("SELECT * FROM products")
for p in cur.fetchall():
    print(f"• #{p['product_id']}: {p['name']:25} | Stock Left: {p['stock']} | Price: ₹{p['price']:,.2f}")

conn.close()

# =========================================================================
# PROJECT 2: WEATHER & FINANCIAL REST API CLIENT
# =========================================================================

class WeatherStockAPIClient:
    """Client for fetching weather and stock market data."""
    
    def __init__(self, api_key="demo_key"):
        self.api_key = api_key

    def get_weather_report(self, city):
        """Simulates fetching real-time weather metrics."""
        simulated_data = {
            "city": city.title(),
            "temperature_c": 28.5,
            "humidity_percent": 65,
            "condition": "Partly Cloudy ⛅",
            "wind_speed_kmh": 14.2
        }
        return (
            f"🌤️ WEATHER REPORT: {simulated_data['city']}\n"
            f"  • Temperature: {simulated_data['temperature_c']}°C\n"
            f"  • Condition:   {simulated_data['condition']}\n"
            f"  • Humidity:    {simulated_data['humidity_percent']}%\n"
            f"  • Wind Speed:  {simulated_data['wind_speed_kmh']} km/h"
        )

    def get_stock_quote(self, ticker):
        """Simulates fetching real-time stock ticker price."""
        simulated_quotes = {
            "TCS": {"price": 4250.00, "change_pct": +1.45},
            "INFY": {"price": 1820.50, "change_pct": -0.80},
            "RELIANCE": {"price": 2980.00, "change_pct": +0.65}
        }
        quote = simulated_quotes.get(ticker.upper(), {"price": 1000.0, "change_pct": 0.0})
        symbol = "🟢 +" if quote["change_pct"] >= 0 else "🔴 "
        return f"📈 [{ticker.upper()}] Stock: ₹{quote['price']:,.2f} ({symbol}{quote['change_pct']}%)"

# Run Client Demonstration:
client = WeatherStockAPIClient()
print("--- 🌦️ Live Weather Client ---")
print(client.get_weather_report("Hyderabad"))
print(client.get_weather_report("Bengaluru"))

print("\n--- 📊 Live Stock Market Quotes ---")
print(client.get_stock_quote("TCS"))
print(client.get_stock_quote("INFY"))

# =========================================================================
# PROJECT 3: SQLALCHEMY 2.0 TASK & PROJECT MANAGEMENT SYSTEM
# =========================================================================
from typing import List, Optional
from sqlalchemy import create_engine, String, Boolean, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class Project(Base):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(50), nullable=False)
    
    # One-to-Many Relationship:
    tasks: Mapped[List["Task"]] = relationship(back_populates="project", cascade="all, delete-orphan")

    def __repr__(self):
        return f""

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    description: Mapped[str] = mapped_column(String(100), nullable=False)
    is_completed: Mapped[bool] = mapped_column(Boolean, default=False)
    project_id: Mapped[int] = mapped_column(ForeignKey("projects.id"))

    project: Mapped["Project"] = relationship(back_populates="tasks")

    def __repr__(self):
        status = "✅ Done" if self.is_completed else "⏳ In Progress"
        return f"  • Task #{self.id}: {self.description:35} [{status}]"

# Initialize SQLite database:
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)

# Execute CRUD operations:
with Session(engine) as session:
    # 1. Create a project with nested tasks:
    p1 = Project(title="ManaCompiler Python Masterclass")
    p1.tasks.append(Task(description="Write Phase 8 Advanced Python Chapters", is_completed=True))
    p1.tasks.append(Task(description="Write Phase 9 Databases and APIs", is_completed=True))
    p1.tasks.append(Task(description="Deploy to Production Server", is_completed=False))

    session.add(p1)
    session.commit()

    # 2. Query Project and print full task list:
    queried_project = session.scalar(select(Project).where(Project.id == 1))
    print(f"--- 📋 {queried_project.title} ---")
    for t in queried_project.tasks:
        print(t)

    # 3. Mark a task as completed:
    stmt = select(Task).where(Task.description.contains("Deploy"))
    pending_task = session.scalar(stmt)
    if pending_task:
        pending_task.is_completed = True
        session.commit()
        print("\n🚀 Updated Task:", pending_task)

# =========================================================================
# PROJECT 4: RESILIENT GITHUB API CLIENT
# =========================================================================

class GitHubAPIClient:
    """Production API Client for GitHub REST API v3."""
    
    BASE_URL = "https://api.github.com"

    def __init__(self, auth_token=None):
        self.headers = {
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "ManaCompiler-App/1.0"
        }
        if auth_token:
            self.headers["Authorization"] = f"token {auth_token}"

    def fetch_repo_summary(self, owner, repo):
        """Fetches and formats repository metadata safely."""
        simulated_response = {
            "full_name": f"{owner}/{repo}",
            "stargazers_count": 34890,
            "forks_count": 4210,
            "open_issues_count": 142,
            "language": "Python",
            "description": "The uncompromising Python code formatter"
        }
        
        return {
            "repository": simulated_response["full_name"],
            "stars": f"{simulated_response['stargazers_count']:,} ⭐",
            "forks": f"{simulated_response['forks_count']:,} 🍴",
            "issues": f"{simulated_response['open_issues_count']} open ⚠️",
            "primary_language": simulated_response["language"],
            "about": simulated_response["description"]
        }

# Run GitHub API Client Demo:
gh = GitHubAPIClient()
repo_info = gh.fetch_repo_summary("psf", "black")

print("--- 🐙 GitHub Repository Metadata ---")
for key, val in repo_info.items():
    print(f"• {key.replace('_', ' ').title():18}: {val}")