Databases: sqlite3 & MySQL

SQL from Python with sqlite3, parameters vs SQL injection, transactions, connecting to MySQL and a peek at ORMs.

Files are fine for small amounts of data, but real applications need to store, query and update data safely, with many users at once. That's what relational databases are for. Python's standard library includes sqlite3, a complete SQL database in a single file — perfect for learning, tests, desktop apps and small sites. For multi-user production systems you'll often use a server such as MySQL (or PostgreSQL). The good news: Python's database API (PEP 249) looks almost the same for all of them.

This lesson assumes basic SQL (CREATE TABLE, SELECT, INSERT, JOIN). If that's new, Elephantoo's MySQL course covers it.

sqlite3: a database with zero setup#

Python

import sqlite3

conn = sqlite3.connect("shop.db")      # creates the file if it doesn't exist
cur = conn.cursor()

cur.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL UNIQUE,
        price REAL NOT NULL CHECK (price >= 0),
        stock INTEGER NOT NULL DEFAULT 0
    )
""")
cur.execute("DELETE FROM products")    # start fresh each run (demo only)

cur.execute("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)", ("Keyboard", 2499, 15))
print("new id:", cur.lastrowid)

conn.commit()                          # make the changes permanent
conn.close()

Output

new id: 1

The pattern is the same for every database:

Connect to get a connection.
Get a cursor and execute SQL.
Commit changes (inserts, updates, deletes).
Close the connection.

Use sqlite3.connect(":memory:") for a throw-away in-memory database — ideal for tests.

Parameters: never format SQL yourself#

Values always go in as parameters with ? placeholders — never by pasting them into the SQL string:

Python

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (name TEXT, is_admin INTEGER)")
conn.executemany("INSERT INTO users VALUES (?, ?)", [("ada", 1), ("bob", 0)])

evil = "nobody' OR '1'='1"

# DANGEROUS: the input becomes part of the SQL
unsafe = conn.execute(f"SELECT name FROM users WHERE name = '{evil}'").fetchall()
print("f-string query returned:", unsafe)

# SAFE: the driver treats the input purely as a value
safe = conn.execute("SELECT name FROM users WHERE name = ?", (evil,)).fetchall()
print("parameterised query returned:", safe)

Output

f-string query returned: [('ada',), ('bob',)]
parameterised query returned: []

That's SQL injection — one of the most common and damaging security bugs in web applications. Parameters prevent it completely. Note the trailing comma in (evil,): parameters must be a tuple (or list), even for one value.

Inserting many rows and querying#

Python

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row                 # rows behave like dicts too
conn.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT, price REAL, stock INTEGER)")

products = [("Keyboard", 2499, 15), ("Mouse", 799, 40), ("Monitor", 12999, 0), ("Webcam", 3499, 7)]
conn.executemany("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)", products)
conn.commit()

cur = conn.execute("SELECT name, price FROM products WHERE stock > ? ORDER BY price DESC", (0,))
for row in cur:                                # iterate the cursor directly
    print(f"{row['name']:<10} ₹{row['price']:>8,.0f}")

one = conn.execute("SELECT * FROM products WHERE name = ?", ("Mouse",)).fetchone()
print(dict(one))
print(tuple(conn.execute("SELECT COUNT(*), SUM(price * stock) FROM products").fetchone()))
missing = conn.execute("SELECT * FROM products WHERE name = ?", ("Tablet",)).fetchone()
print(missing)

Output

Webcam     ₹   3,499
Keyboard   ₹   2,499
Mouse      ₹     799
{'id': 2, 'name': 'Mouse', 'price': 799.0, 'stock': 40}
(4, 93938.0)
None

fetchone() returns the next row or None; fetchall() returns a list; fetchmany(n) returns up to n rows. Iterating the cursor is the memory-friendly option.
Setting conn.row_factory = sqlite3.Row lets you access columns by name.

Transactions: all or nothing#

A transaction groups several statements so that either all of them take effect or none do. Using the connection as a context manager commits on success and rolls back if an exception occurs:

Python

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE accounts (name TEXT PRIMARY KEY, balance INTEGER CHECK (balance >= 0))")
with conn:
    conn.executemany("INSERT INTO accounts VALUES (?, ?)", [("ada", 1000), ("bob", 200)])


def transfer(conn, src, dst, amount):
    try:
        with conn:                                # one transaction
            conn.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", (amount, dst))
            conn.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?", (amount, src))
        print(f"moved {amount} from {src} to {dst}")
    except sqlite3.IntegrityError as e:
        print("transfer failed, rolled back:", e)


transfer(conn, "ada", "bob", 300)
transfer(conn, "bob", "ada", 9999)                # would make bob negative
print(conn.execute("SELECT * FROM accounts ORDER BY name").fetchall())
conn.close()

Output

moved 300 from ada to bob
transfer failed, rolled back: CHECK constraint failed: balance >= 0
[('ada', 700), ('bob', 500)]

The failed transfer had already credited ada before the debit failed — the rollback undid it, so no money was created out of thin air. Note that with conn: manages the transaction but doesn't close the connection; call close() (or use contextlib.closing).

JOINs and a small repository layer#

In real code, keep SQL in a few well-named functions instead of scattering it everywhere:

Python

import sqlite3
from dataclasses import dataclass


@dataclass
class OrderSummary:
    customer: str
    orders: int
    total: float


SCHEMA = """
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT NOT NULL);
CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL REFERENCES customers(id),
    amount REAL NOT NULL
);
"""


def setup(conn):
    conn.executescript(SCHEMA)
    with conn:
        conn.executemany("INSERT INTO customers (id, name) VALUES (?, ?)", [(1, "Ada"), (2, "Grace"), (3, "Linus")])
        conn.executemany("INSERT INTO orders (customer_id, amount) VALUES (?, ?)",
                         [(1, 2499), (1, 799), (2, 12999), (1, 349)])


def order_summaries(conn) -> list[OrderSummary]:
    rows = conn.execute("""
        SELECT c.name, COUNT(o.id), COALESCE(SUM(o.amount), 0)
        FROM customers c LEFT JOIN orders o ON o.customer_id = c.id
        GROUP BY c.id ORDER BY 3 DESC
    """)
    return [OrderSummary(*row) for row in rows]


conn = sqlite3.connect(":memory:")
setup(conn)
for s in order_summaries(conn):
    print(s)

Output

OrderSummary(customer='Grace', orders=1, total=12999.0)
OrderSummary(customer='Ada', orders=3, total=3647.0)
OrderSummary(customer='Linus', orders=0, total=0)

Connecting to MySQL#

For MySQL 8.x, install the official driver into your virtual environment:

Terminal

pip install mysql-connector-python

Then create a database and a dedicated user (in the mysql client, as an admin):

SQL

CREATE DATABASE shop;
CREATE USER 'shop_app'@'localhost' IDENTIFIED BY 'change-me';
GRANT ALL PRIVILEGES ON shop.* TO 'shop_app'@'localhost';

The Python code follows the same connect → cursor → execute → commit pattern. The main differences: the placeholder is %s (for every type), and connection details come from environment variables rather than being hard-coded:

mysql_demo.py

import os

import mysql.connector
from mysql.connector import errorcode

config = {
    "host": os.environ.get("DB_HOST", "localhost"),
    "user": os.environ.get("DB_USER", "shop_app"),
    "password": os.environ.get("DB_PASSWORD", "change-me"),
    "database": os.environ.get("DB_NAME", "shop"),
}

try:
    conn = mysql.connector.connect(**config)
except mysql.connector.Error as err:
    if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
        raise SystemExit("bad username or password")
    raise

try:
    cur = conn.cursor(dictionary=True)          # rows as dicts
    cur.execute("DROP TABLE IF EXISTS products")
    cur.execute("""
        CREATE TABLE products (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(100) NOT NULL UNIQUE,
            price DECIMAL(10, 2) NOT NULL,
            stock INT NOT NULL DEFAULT 0
        )
    """)
    cur.executemany(
        "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)",
        [("Keyboard", 2499, 15), ("Mouse", 799, 40), ("Monitor", 12999, 0)],
    )
    conn.commit()                               # autocommit is off by default
    print("inserted", cur.rowcount, "rows")

    cur.execute("SELECT name, price FROM products WHERE stock > %s ORDER BY price", (0,))
    for row in cur.fetchall():
        print(row["name"], row["price"])
finally:
    conn.close()

Output

inserted 3 rows
Mouse 799.00
Keyboard 2499.00

Notice that DECIMAL columns come back as Python Decimal objects — exact, which is what you want for money. Other popular drivers are PyMySQL (pure Python, same %s style) and, for async code, aiomysql. For PostgreSQL, the equivalent is psycopg.

Connection pooling and ORMs

Opening a connection per request is slow. Web apps use a connection pool (mysql.connector.pooling.MySQLConnectionPool, or the pool built into an ORM). Many teams also use an ORM (Object-Relational Mapper) such as SQLAlchemy or the Django ORM, which maps tables to Python classes and generates SQL for you:

Python

# SQLAlchemy 2.x style (pip install sqlalchemy) — a preview, not needed for this course
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)
    price: Mapped[float]


engine = create_engine("sqlite:///:memory:")    # or "mysql+mysqlconnector://user:pw@host/shop"
Base.metadata.create_all(engine)
with Session(engine) as session:
    session.add_all([Product(name="Pen", price=25), Product(name="Ink", price=60)])
    session.commit()
    cheap = session.query(Product).filter(Product.price < 50).all()
    print([p.name for p in cheap])

Output

['Pen']

ORMs speed up development, but learn plain SQL first — you'll need it to understand and debug what the ORM does.

Common mistakes#

Building SQL with f-strings or + — SQL injection. Always use parameters.
Forgetting to commit() — your inserts vanish.
Forgetting the comma in one-element parameter tuples: (name,), not (name).
Mixing up placeholders: ? for sqlite3, %s for MySQL drivers.
Hard-coding passwords in code — use environment variables or a secrets manager.
Using floats for money columns — use DECIMAL in MySQL and Decimal in Python (or integer paise).
Leaving connections open — close them, or use a pool.

What's next#

Once data is in a database or CSV, you'll want to analyse it. Next up: data analysis with NumPy and pandas.

sqlite3: a database with zero setup#

Parameters: never format SQL yourself#

Inserting many rows and querying#

Transactions: all or nothing#

JOINs and a small repository layer#

Connecting to MySQL#

Connection pooling and ORMs

Common mistakes#

What's next#

Quick quiz