Databases & Storage

import psycopg2

def transfer_money(from_account, to_account, amount):
    """Transfer money between accounts with ACID guarantees."""
    conn = psycopg2.connect(
        host="localhost", database="bank",
        user="admin", password="secret"
    )

    try:
        # Start transaction (Atomicity: all or nothing)
        cursor = conn.cursor()

        # Check sufficient balance (Consistency)
        cursor.execute(
            "SELECT balance FROM accounts WHERE id = %s FOR UPDATE",
            (from_account,)
        )
        balance = cursor.fetchone()[0]

        if balance < amount:
            raise ValueError("Insufficient funds")

        # Debit sender
        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_account)
        )

        # Credit receiver
        cursor.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_account)
        )

        # Record the transaction
        cursor.execute(
            "INSERT INTO transfers (from_id, to_id, amount) VALUES (%s, %s, %s)",
            (from_account, to_account, amount)
        )

        # Commit (Durability: persisted to disk)
        conn.commit()
        print(f"Transferred ${amount} successfully")

    except Exception as e:
        # Rollback on any error (Atomicity)
        conn.rollback()
        print(f"Transfer failed: {e}")
    finally:
        conn.close()

-- Without index: Full table scan O(n)
-- With 10M rows, this scans ALL 10M rows
SELECT * FROM users WHERE email = '[email protected]';

-- Create an index on the email column
CREATE INDEX idx_users_email ON users(email);

-- Now the same query uses O(log n) lookup!
-- With 10M rows: ~24 comparisons vs 10,000,000

-- Composite index for multi-column queries
CREATE INDEX idx_orders_user_date
ON orders(user_id, created_at DESC);

-- This query benefits from the composite index
SELECT * FROM orders
WHERE user_id = 42
ORDER BY created_at DESC
LIMIT 10;

-- Check query execution plan
EXPLAIN ANALYZE
SELECT * FROM users WHERE email = '[email protected]';
-- Without index: Seq Scan, cost=0..1234, time=450ms
-- With index:    Index Scan, cost=0..8, time=0.1ms

class ShardRouter:
    """Route queries to the correct database shard."""

    def __init__(self, shard_connections):
        self.shards = shard_connections  # List of DB connections
        self.num_shards = len(shard_connections)

    def get_shard(self, user_id):
        """Determine which shard holds this user's data."""
        shard_index = user_id % self.num_shards
        return self.shards[shard_index]

    def get_user(self, user_id):
        """Fetch a user from the correct shard."""
        shard = self.get_shard(user_id)
        cursor = shard.cursor()
        cursor.execute(
            "SELECT * FROM users WHERE id = %s",
            (user_id,)
        )
        return cursor.fetchone()

    def create_user(self, user_id, name, email):
        """Insert a user into the correct shard."""
        shard = self.get_shard(user_id)
        cursor = shard.cursor()
        cursor.execute(
            "INSERT INTO users (id, name, email) VALUES (%s, %s, %s)",
            (user_id, name, email)
        )
        shard.commit()

    def query_all_shards(self, query):
        """Fan-out query to all shards (expensive!)."""
        results = []
        for shard in self.shards:
            cursor = shard.cursor()
            cursor.execute(query)
            results.extend(cursor.fetchall())
        return results

# Usage
router = ShardRouter([db_shard_0, db_shard_1, db_shard_2, db_shard_3])
user = router.get_user(42)  # Routes to shard 42 % 4 = shard 2