How SQL Helps You Land a Python Developer Job Faster

import sqlite3

conn = sqlite3.connect("shop.db")
conn.row_factory = sqlite3.Row   # access columns by name
cur = conn.cursor()

cur.execute("""
CREATE TABLE IF NOT EXISTS customers (
    id    INTEGER PRIMARY KEY,
    name  TEXT NOT NULL,
    city  TEXT
)""")

cur.executemany(
    "INSERT INTO customers (name, city) VALUES (?, ?)",
    [("Asha", "Noida"), ("Ravi", "Delhi"), ("Meera", "Noida")],
)
conn.commit()

city = "Noida"
for row in cur.execute("SELECT id, name FROM customers WHERE city = ?", (city,)):
    print(row["id"], row["name"])

conn.close()

$ python3 basics.py
1 Asha
3 Meera

# sqlite3 ships with Python, so you can practise
# SQL without installing a database server.
# The same DB-API pattern (connect, cursor,
# execute, commit) works with psycopg for
# PostgreSQL and other drivers, though the
# placeholder style may differ (%s instead of ?).

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT, is_admin INTEGER)")
conn.executemany("INSERT INTO users (email, is_admin) VALUES (?, ?)",
                 [("a@x.com", 0), ("boss@x.com", 1)])

user_input = "nobody' OR '1'='1"

# UNSAFE: input is pasted into the SQL text
unsafe = f"SELECT email FROM users WHERE email = '{user_input}'"
print("unsafe:", conn.execute(unsafe).fetchall())

# SAFE: the driver sends the value separately
safe = "SELECT email FROM users WHERE email = ?"
print("safe:  ", conn.execute(safe, (user_input,)).fetchall())

# Output:
# unsafe: [('a@x.com',), ('boss@x.com',)]
# safe:   []

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, "
             "balance INTEGER CHECK (balance >= 0))")
conn.executemany("INSERT INTO accounts (id, balance) VALUES (?, ?)", [(1, 500), (2, 100)])
conn.commit()

def transfer(conn, src, dst, amount):
    try:
        with conn:  # commit on success, rollback on exception
            conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, src))
            conn.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, dst))
        return True
    except sqlite3.IntegrityError:
        return False

print(transfer(conn, 1, 2, 200))   # True
print(transfer(conn, 2, 1, 9999))  # False: CHECK fails, nothing changes
print(conn.execute("SELECT id, balance FROM accounts").fetchall())
# [(1, 300), (2, 300)]

# pip install sqlalchemy   (2.0 style)
from sqlalchemy import create_engine, ForeignKey, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class Customer(Base):
    __tablename__ = "customers"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    orders: Mapped[list["Order"]] = relationship(back_populates="customer")

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    amount: Mapped[int]
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    customer: Mapped["Customer"] = relationship(back_populates="orders")

engine = create_engine("sqlite:///shop_orm.db", echo=True)  # echo prints SQL
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Customer(name="Asha", orders=[Order(amount=1200), Order(amount=800)]))
    session.commit()

    stmt = (select(Customer.name, func.sum(Order.amount).label("total"))
            .join(Customer.orders)
            .group_by(Customer.name))
    for name, total in session.execute(stmt):
        print(name, total)          # Asha 2000

from sqlalchemy import select, text
from sqlalchemy.orm import Session, selectinload
# (continues from the ORM Models tab)

# N+1: one query for customers, then one per customer for orders
with Session(engine) as s:
    for c in s.scalars(select(Customer)):
        print(c.name, len(c.orders))

# Fix: load all orders in one extra query
with Session(engine) as s:
    stmt = select(Customer).options(selectinload(Customer.orders))
    for c in s.scalars(stmt):
        print(c.name, len(c.orders))

# Raw SQL when it is clearer, still parameterised
report = text("""
    SELECT c.name, COUNT(o.id) AS order_count, SUM(o.amount) AS total
    FROM customers c
    LEFT JOIN orders o ON o.customer_id = c.id
    GROUP BY c.name
    HAVING SUM(o.amount) > :min_total
    ORDER BY total DESC
""")
with engine.connect() as conn:
    for row in conn.execute(report, {"min_total": 500}):
        print(row.name, row.order_count, row.total)

-- schema.sql  (works in SQLite and PostgreSQL)
CREATE TABLE customers (
    id      INTEGER PRIMARY KEY,
    name    TEXT NOT NULL,
    city    TEXT,
    joined  DATE
);

CREATE TABLE orders (
    id           INTEGER PRIMARY KEY,
    customer_id  INTEGER NOT NULL REFERENCES customers(id),
    amount       INTEGER NOT NULL,
    created      DATE NOT NULL
);

CREATE INDEX idx_orders_customer ON orders (customer_id);

-- Load it from Python:
-- conn.executescript(open("schema.sql").read())

QUERY DRILLS (write each in SQL, then run from Python)

1. Customers from Noida, newest first
2. Total order amount per customer (GROUP BY)
3. Customers with no orders (LEFT JOIN ... IS NULL)
4. Customers whose total is above a value (HAVING)
5. Latest order per customer (ROW_NUMBER() OVER
   (PARTITION BY customer_id ORDER BY created DESC))
6. Monthly revenue (strftime / date_trunc)
7. Top 3 customers by spend (ORDER BY ... LIMIT 3)
8. Same as 2, but written with SQLAlchemy select()
9. Insert 1,000 rows with executemany in one transaction
10. Explain query 2 with EXPLAIN QUERY PLAN / EXPLAIN

PYTHON + SQL PORTFOLIO PROJECT CHECKLIST

[ ] Small API (Flask or FastAPI) over 2-3 related tables
[ ] Schema with primary keys, foreign keys, one index
[ ] All queries parameterised - no f-strings in SQL
[ ] At least one multi-step write in a transaction
[ ] One endpoint using a JOIN or eager loading
[ ] One report query with GROUP BY or a window function
[ ] Migrations (Alembic or Django migrations)
[ ] Tests that use a temporary SQLite database
[ ] README: schema diagram + example queries
[ ] Pushed to GitHub with clear commits