Complete SQL reference covering MySQL, PostgreSQL, SQLite, Oracle, and SQL Server — DQL, DML, DDL, DCL, aggregate functions, window functions, transactions, indexes, and Python integration.
Part of The Direct Path to Linux Ubuntu — free for all.
SQLite3 is built into Python — no install needed. For PostgreSQL: pip install psycopg2-binary. For MySQL: pip install mysql-connector-python.
Most-used SQL statements in one place.
-- SQL Quick Reference — works on MySQL, PostgreSQL, SQLite
-- ── DQL (Query) ──────────────────────────────────────────────
SELECT col1, col2 FROM table; -- basic
SELECT DISTINCT col FROM table; -- unique
SELECT * FROM t WHERE col='val' AND n > 5; -- filter
SELECT * FROM t ORDER BY col DESC LIMIT 10 OFFSET 20; -- sort+page
SELECT a.*, b.name FROM a JOIN b ON a.bid = b.id; -- join
SELECT col, COUNT(*) FROM t GROUP BY col HAVING COUNT(*) > 1;
-- ── DML (Modify data) ────────────────────────────────────────
INSERT INTO t (col1,col2) VALUES ('a', 42);
INSERT INTO t (col1,col2) VALUES ('a',1),('b',2),('c',3); -- bulk
UPDATE t SET col1='new', col2=col2+1 WHERE id = 5;
DELETE FROM t WHERE id = 5;
-- ── DDL (Structure) ──────────────────────────────────────────
CREATE TABLE t (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
score INT DEFAULT 0 CHECK (score BETWEEN 0 AND 100),
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
ALTER TABLE t ADD COLUMN email VARCHAR(150) UNIQUE;
ALTER TABLE t DROP COLUMN old_col;
CREATE INDEX idx ON t(name);
DROP TABLE IF EXISTS old_table;
-- ── Transactions ─────────────────────────────────────────────
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
COMMIT; -- or ROLLBACK;
-- ── Useful patterns ───────────────────────────────────────────
SELECT COALESCE(col, 'default') FROM t; -- null safe
SELECT CASE WHEN x>0 THEN 'pos' ELSE 'neg' END FROM t;
WITH cte AS (SELECT ...) SELECT * FROM cte; -- CTE
SELECT * FROM t WHERE id IN (SELECT id FROM t2); -- subquery
SELECT * FROM t WHERE NOT EXISTS (SELECT 1 FROM t2 WHERE t2.id=t.id);Create, populate, and query a complete database.
import sqlite3, json
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
# Schema
conn.executescript("""
CREATE TABLE departments (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE
);
CREATE TABLE employees (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
dept_id INTEGER REFERENCES departments(id),
salary REAL NOT NULL,
level TEXT DEFAULT 'mid'
);
CREATE TABLE projects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
budget REAL
);
CREATE TABLE assignments (
emp_id INTEGER REFERENCES employees(id),
proj_id INTEGER REFERENCES projects(id),
role TEXT,
PRIMARY KEY (emp_id, proj_id)
);
""")
# Data
conn.executemany("INSERT INTO departments VALUES (?,?)",
[(1,'Engineering'),(2,'Marketing'),(3,'Data Science')])
conn.executemany("INSERT INTO employees (name,dept_id,salary,level) VALUES (?,?,?,?)", [
('Alice',1,95000,'senior'),('Bob',1,87000,'mid'),
('Carol',2,72000,'mid'), ('Dave',3,110000,'senior'),
('Eve', 3,98000,'senior'),('Frank',1,68000,'junior'),
])
conn.executemany("INSERT INTO projects (title,budget) VALUES (?,?)", [
('Platform Rewrite',500000),('ML Pipeline',200000),('Website Redesign',80000),
])
conn.executemany("INSERT INTO assignments VALUES (?,?,?)",
[(1,1,'lead'),(2,1,'dev'),(6,1,'dev'),(4,2,'lead'),(5,2,'dev'),(3,3,'lead')])
conn.commit()
# Queries
print("=== Salary by department ===")
for r in conn.execute('''
SELECT d.name, COUNT(*) n, AVG(e.salary) avg, MAX(e.salary) top
FROM employees e JOIN departments d ON e.dept_id=d.id
GROUP BY d.id ORDER BY avg DESC'''):
print(f" {r['name']:15} n={r['n']} avg=${r['avg']:,.0f} top=${r['top']:,.0f}")
print("\n=== Senior employees and their projects ===")
for r in conn.execute('''
SELECT e.name, e.salary, d.name dept, p.title project, a.role
FROM employees e
JOIN departments d ON e.dept_id = d.id
LEFT JOIN assignments a ON e.id = a.emp_id
LEFT JOIN projects p ON a.proj_id = p.id
WHERE e.level = 'senior'
ORDER BY e.salary DESC'''):
proj = r['project'] or 'Unassigned'
print(f" {r['name']:8} ${r['salary']:>9,.0f} {r['dept']:15} {proj}")
conn.close()
# python3 db_demo.pyConnect to PostgreSQL and run queries safely.
# pip install psycopg2-binary
import psycopg2
import psycopg2.extras
# Connection — use environment variables in production!
DSN = "postgresql://postgres:password@localhost:5432/mywebuniversity"
def run_demo():
with psycopg2.connect(DSN) as conn:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
# Create table
cur.execute("""
CREATE TABLE IF NOT EXISTS courses (
id SERIAL PRIMARY KEY,
code TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
credits INT NOT NULL DEFAULT 3,
tags TEXT[] DEFAULT '{}',
metadata JSONB DEFAULT '{}'
)
""")
# Bulk insert with execute_values (fast!)
data = [
("CS101","Python Fundamentals",3,["python","beginner"],'{"free":true}'),
("CS201","Go Programming", 3,["go","intermediate"],'{"free":true}'),
("CS301","Data Science with R",4,["r","data-science"],'{"free":true}'),
]
psycopg2.extras.execute_values(cur, """
INSERT INTO courses (code,title,credits,tags,metadata)
VALUES %s ON CONFLICT (code) DO NOTHING
""", data)
# Query with dict results
cur.execute("SELECT * FROM courses ORDER BY credits DESC, code")
for row in cur.fetchall():
print(f" {row['code']} | {row['title']} | tags={row['tags']}")
# Array operations (PostgreSQL native)
cur.execute(
"SELECT title FROM courses WHERE %s = ANY(tags)",
("python",)
)
print("Python courses:", [r['title'] for r in cur.fetchall()])
# JSONB query
cur.execute("SELECT title FROM courses WHERE metadata @> %s", ('{"free":true}',))
print("Free courses:", [r['title'] for r in cur.fetchall()])
# Cleanup
cur.execute("DROP TABLE IF EXISTS courses")
conn.commit()
try:
run_demo()
except psycopg2.OperationalError as e:
print(f"Cannot connect to PostgreSQL: {e}")
print("Start PostgreSQL: sudo service postgresql start")
# python3 pg_demo.pyConnect to MySQL and use parameterized queries.
# pip install mysql-connector-python
import mysql.connector
from mysql.connector import Error
config = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'your_password',
'database': 'mywebuniversity',
'charset': 'utf8mb4',
}
def run_demo():
try:
conn = mysql.connector.connect(**config)
cursor = conn.cursor(dictionary=True) # dict rows
# Create table
cursor.execute("""
CREATE TABLE IF NOT EXISTS students (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(150) NOT NULL UNIQUE,
score DECIMAL(5,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")
# Parameterized insert (prevents SQL injection!)
data = [
('Alice Smith', 'alice@mwu.com', 95.0),
('Bob Jones', 'bob@mwu.com', 87.5),
('Carol Davis', 'carol@mwu.com', 92.0),
]
cursor.executemany(
"INSERT IGNORE INTO students (name,email,score) VALUES (%s,%s,%s)",
data
)
conn.commit()
print(f"Inserted {cursor.rowcount} rows")
# Select with filter
cursor.execute("SELECT * FROM students WHERE score >= %s ORDER BY score DESC", (90,))
for row in cursor.fetchall():
print(f" [{row['id']}] {row['name']} — {row['score']}")
# Update
cursor.execute("UPDATE students SET score = score + 2 WHERE score < 90")
conn.commit()
print(f"Updated {cursor.rowcount} rows")
# Aggregates
cursor.execute("SELECT COUNT(*) n, AVG(score) avg, MAX(score) top FROM students")
stats = cursor.fetchone()
print(f"Stats: n={stats['n']} avg={stats['avg']:.2f} top={stats['top']}")
# Cleanup
cursor.execute("DROP TABLE IF EXISTS students")
conn.commit()
except Error as e:
print(f"MySQL Error: {e}")
finally:
if conn.is_connected():
cursor.close(); conn.close()
print("Connection closed.")
run_demo()
# python3 mysql_demo.pyDefine models, create tables, and query with SQLAlchemy.
# pip install sqlalchemy
from sqlalchemy import (create_engine, Column, Integer, String,
Float, DateTime, ForeignKey, func, text)
from sqlalchemy.orm import DeclarativeBase, relationship, Session
from datetime import datetime
# ── Models ────────────────────────────────────────────────────
class Base(DeclarativeBase):
pass
class Department(Base):
__tablename__ = "departments"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False, unique=True)
employees = relationship("Employee", back_populates="department")
def __repr__(self): return f"<Department {self.name}>"
class Employee(Base):
__tablename__ = "employees"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
email = Column(String(150), nullable=False, unique=True)
salary = Column(Float, nullable=False)
dept_id = Column(Integer, ForeignKey("departments.id"))
created = Column(DateTime, default=datetime.utcnow)
department = relationship("Department", back_populates="employees")
def __repr__(self): return f"<Employee {self.name} ${self.salary:,.0f}>"
# ── Setup ─────────────────────────────────────────────────────
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
with Session(engine) as session:
# Insert
eng = Department(id=1, name="Engineering")
data = Department(id=2, name="Data Science")
session.add_all([eng, data])
session.flush()
session.add_all([
Employee(name="Alice", email="a@mwu.com", salary=95000, dept_id=1),
Employee(name="Bob", email="b@mwu.com", salary=87000, dept_id=1),
Employee(name="Carol", email="c@mwu.com", salary=110000,dept_id=2),
Employee(name="Dave", email="d@mwu.com", salary=98000, dept_id=2),
])
session.commit()
# ORM queries
top = session.query(Employee).filter(Employee.salary > 90000) .order_by(Employee.salary.desc()).all()
print("High earners:", top)
# Join
results = session.query(Employee.name, Department.name, Employee.salary) .join(Department) .order_by(Employee.salary.desc()) .all()
for name, dept, salary in results:
print(f" {name:8} {dept:15} ${salary:,.0f}")
# Aggregates
stats = session.query(
Department.name,
func.count(Employee.id).label("count"),
func.avg(Employee.salary).label("avg_salary")
).join(Employee).group_by(Department.name).all()
for dept, count, avg in stats:
print(f" {dept:15} n={count} avg=${avg:,.0f}")
print("Done!")
# python3 sqlalchemy_demo.py