Python connects to databases using the DB-API 2.0 interface (PEP 249). Built-in sqlite3 needs no install. psycopg2 or asyncpg connects to PostgreSQL. mysql-connector-python connects to MySQL. SQLAlchemy is the ORM layer that works with all databases.
| Statement / Feature | Syntax | Description |
|---|---|---|
| sqlite3.connect | sqlite3.connect(':memory:') / sqlite3.connect('file.db') | Connect to SQLite (built-in) |
| conn.cursor | cursor = conn.cursor() | Create cursor for executing SQL |
| cursor.execute | cursor.execute(sql, params) | Execute SQL with parameter binding |
| cursor.executemany | cursor.executemany(sql, list_of_params) | Execute SQL for multiple parameter sets |
| cursor.fetchone | cursor.fetchone() → tuple | Fetch next row as tuple |
| cursor.fetchall | cursor.fetchall() → list of tuples | Fetch all rows |
| cursor.fetchmany | cursor.fetchmany(size) → list | Fetch up to size rows |
| cursor.rowcount | cursor.rowcount → int | Rows affected by last DML statement |
| conn.commit | conn.commit() | Commit transaction |
| conn.rollback | conn.rollback() | Roll back transaction |
| conn.close | conn.close() | Close connection |
| Row factory | conn.row_factory = sqlite3.Row | Enable dict-like row access by column name |
| Placeholders sqlite3 | cursor.execute(sql, (val1, val2)) | SQLite: ? placeholders |
| Placeholders psycopg2 | cursor.execute(sql, (val1, val2)) | psycopg2: %s placeholders |
| psycopg2.connect | psycopg2.connect(dsn) / psycopg2.connect(host=,dbname=,user=,password=) | PostgreSQL connection |
| RealDictCursor | psycopg2.extras.RealDictCursor | psycopg2 cursor returning dicts |
| execute_values | psycopg2.extras.execute_values(cursor, sql, data) | Fast bulk insert for PostgreSQL |
| mysql.connector | mysql.connector.connect(host=,database=,user=,password=) | MySQL connection |
| context manager | with sqlite3.connect('db') as conn: | Auto-commit on success, rollback on exception |
| SQLAlchemy engine | create_engine('postgresql://user:pass@host/db') | SQLAlchemy database URL |
| SQLAlchemy session | Session = sessionmaker(bind=engine); session = Session() | ORM session |
| SQLAlchemy query | session.query(Model).filter(Model.col==val).all() | ORM query |
| SQLAlchemy ORM | class Model(Base): __tablename__ = 'table' | ORM model definition |
import sqlite3
import json
from pathlib import Path
from contextlib import contextmanager
DB_PATH = "mywebuniversity.db"
# ── 1. Context manager for connection ────────────────────────
@contextmanager
def get_db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row # dict-like access
conn.execute("PRAGMA foreign_keys = ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
# ── 2. Create schema ─────────────────────────────────────────
def create_schema():
with get_db() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS courses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
credits INTEGER NOT NULL DEFAULT 3,
price REAL NOT NULL DEFAULT 0.0
);
CREATE TABLE IF NOT EXISTS students (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL UNIQUE,
enrolled TEXT NOT NULL DEFAULT (date('now'))
);
CREATE TABLE IF NOT EXISTS enrollments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
student_id INTEGER NOT NULL REFERENCES students(id),
course_id INTEGER NOT NULL REFERENCES courses(id),
score REAL,
UNIQUE(student_id, course_id)
);
CREATE INDEX IF NOT EXISTS idx_enroll_student ON enrollments(student_id);
CREATE INDEX IF NOT EXISTS idx_enroll_course ON enrollments(course_id);
""")
print("Schema created.")
# ── 3. Insert data ────────────────────────────────────────────
def seed_data():
courses = [("CS101","Python Fundamentals",3,0),
("CS201","Go Programming",3,0),
("CS301","Data Science with R",4,0)]
students = [("Alice Smith","alice@mwu.com"),
("Bob Jones", "bob@mwu.com"),
("Carol Davis","carol@mwu.com")]
with get_db() as conn:
conn.executemany(
"INSERT OR IGNORE INTO courses(code,title,credits,price) VALUES(?,?,?,?)",
courses)
conn.executemany(
"INSERT OR IGNORE INTO students(name,email) VALUES(?,?)",
students)
# Enroll all students in CS101
conn.execute("""
INSERT OR IGNORE INTO enrollments(student_id,course_id,score)
SELECT s.id, c.id, NULL
FROM students s, courses c WHERE c.code='CS101'
""")
print("Data seeded.")
# ── 4. Query ──────────────────────────────────────────────────
def get_course_stats():
with get_db() as conn:
rows = conn.execute("""
SELECT c.code, c.title,
COUNT(e.id) AS enrolled,
AVG(e.score) AS avg_score,
MAX(e.score) AS top_score
FROM courses c
LEFT JOIN enrollments e ON c.id = e.course_id
GROUP BY c.id, c.code, c.title
ORDER BY enrolled DESC
""").fetchall()
for r in rows:
avg = f"{r['avg_score']:.1f}" if r['avg_score'] else "N/A"
print(f" {r['code']} | {r['title'][:30]} | n={r['enrolled']} avg={avg}")
# ── 5. Parameterized update ───────────────────────────────────
def update_score(student_email: str, course_code: str, score: float):
with get_db() as conn:
cur = conn.execute("""
UPDATE enrollments
SET score = ?
WHERE student_id = (SELECT id FROM students WHERE email=?)
AND course_id = (SELECT id FROM courses WHERE code=?)
""", (score, student_email, course_code))
print(f"Updated {cur.rowcount} row(s).")
if __name__ == "__main__":
create_schema()
seed_data()
update_score("alice@mwu.com", "CS101", 95.5)
update_score("bob@mwu.com", "CS101", 87.0)
get_course_stats()
Path(DB_PATH).unlink()
# python3 sql_demo.py (sqlite3 is built-in — no install needed)