Working with Databases from Code

How an application talks to a database

A web shop does not type Structured Query Language (SQL) into a console. Its code sends SQL to the database server over the network (or, for SQLite, to a file on the same machine) and turns the answer back into values the program can use. Whatever the language or database, the steps are the same:

1. Load a driver. A database driver is the library that knows one database's network protocol and translates between it and the programming language: sqlite3 (built into Python), psycopg for PostgreSQL, mysql-connector-python for MySQL, pyodbc for SQL Server through Open Database Connectivity (ODBC). In Java the standard interface is Java Database Connectivity (JDBC); in .NET it is ADO.NET.
2. Open a connection. The driver connects to the server and logs in. A connection is a session: it has a user, a current transaction and settings. Opening one takes time (network round trips, authentication, often Transport Layer Security (TLS)), so programs reuse connections.
3. Create a cursor and execute a query. A cursor is the object that sends a statement over the connection and keeps track of its result, like a bookmark moving through the rows.
4. Fetch the rows one at a time, in batches, or all at once.
5. Commit changes (or roll them back) and close the connection, or return it to a pool.

Connection strings and credentials

The driver needs to know where the database is and how to log in. A connection string packs that into one text value: host, port, database name, user, and options such as encryption. Formats differ per driver:

PostgreSQL URL:   postgresql://shop_app:PASSWORD@db.example.internal:5432/shop?sslmode=verify-full
SQL Server:       Server=db.example.internal,1433;Database=shop;User Id=shop_app;Password=PASSWORD;Encrypt=True
SQLite (Python):  shop.db

A connection string often contains a password, which makes it a secret. Never write it into the source code: code ends up in version control, in backups and on colleagues' laptops, and a leaked database password is one of the most damaging leaks there is. Instead, the program reads it at startup from an environment variable set by the deployment, or fetches it from a secrets manager (see Cloud Security and Compliance):

import os

import psycopg  # PostgreSQL driver, installed with pip

# The deployment sets DATABASE_URL; the code never contains the password.
with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
    ...

The database account in that connection string should have only the permissions the application needs: reading and writing its own tables, not dropping them or creating users (see Database Security).

Python's sqlite3, end to end

Python's built-in sqlite3 module follows the Python database interface (DB-API, Python Enhancement Proposal 249), so the same pattern works with other drivers. This example walks through all the steps: connect, create a table, insert with parameters, query, fetch, close.

import sqlite3

# 1. Connect (":memory:" = a temporary database; normally a file path)
conn = sqlite3.connect(":memory:")

# 2. Create a table
conn.execute("""
    CREATE TABLE products (
        product_id INTEGER PRIMARY KEY,
        name       TEXT    NOT NULL,
        category   TEXT    NOT NULL,
        price      INTEGER NOT NULL CHECK (price > 0)
    )
""")

# 3. Insert with parameters: the values travel separately from the SQL
conn.execute(
    "INSERT INTO products (name, category, price) VALUES (?, ?, ?)",
    ("USB-C cable 2 m", "Cables", 9),
)
conn.executemany(
    "INSERT INTO products (name, category, price) VALUES (?, ?, ?)",
    [
        ("Wireless mouse", "Peripherals", 25),
        ("27-inch monitor", "Displays", 249),
        ("HDMI cable 3 m", "Cables", 12),
    ],
)
conn.commit()

# 4. Query and fetch: each row is a tuple
cursor = conn.execute(
    "SELECT name, price FROM products WHERE category = ? ORDER BY price",
    ("Cables",),
)
print(cursor.fetchall())

# 5. Rows that behave like dictionaries
conn.row_factory = sqlite3.Row
row = conn.execute("SELECT * FROM products WHERE product_id = ?", (3,)).fetchone()
print(row["name"], row["price"])
print(dict(row))

# 6. Close the connection
conn.close()

[('USB-C cable 2 m', 9), ('HDMI cable 3 m', 12)]
27-inch monitor 249
{'product_id': 3, 'name': '27-inch monitor', 'category': 'Displays', 'price': 249}

Step by step:

• conn.execute(...) is a shortcut that creates a cursor, executes the statement and returns the cursor. With other drivers you write cur = conn.cursor() and cur.execute(...).
• executemany runs one statement for every tuple in the list, which is faster than a loop of separate calls.
• conn.commit() makes the inserts permanent. Without it, closing the connection would discard them.
• fetchall() returns a list of tuples; fetchone() returns the next row or None; fetchmany(100) returns a batch. For large results, loop over the cursor instead of fetching everything into memory.
• With row_factory = sqlite3.Row each row can be read by column name, which keeps code correct when someone adds a column to the query. (Note that ("Cables",) needs the comma: it is a tuple with one value.)

Parameterized queries, always

Every value that comes from outside the code (a form field, a URL, a file, another system) goes into the query as a parameter. The SQL text contains only a placeholder; the driver sends the value separately, and the database treats it strictly as data. That is the primary defense against SQL injection for values, and it also handles quotes in names such as O'Brien correctly.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE customers (customer_id INTEGER PRIMARY KEY, last_name TEXT, city TEXT)")
conn.executemany(
    "INSERT INTO customers (last_name, city) VALUES (?, ?)",
    [("de Vries", "Utrecht"), ("O'Brien", "Dublin"), ("Jansen", "Utrecht")],
)

# Positional placeholders: values in the same order as the question marks
rows = conn.execute(
    "SELECT last_name FROM customers WHERE city = ? AND customer_id > ?",
    ("Utrecht", 1),
).fetchall()
print(rows)

# Named placeholders: values from a dictionary
rows = conn.execute(
    "SELECT customer_id, city FROM customers WHERE last_name = :name",
    {"name": "O'Brien"},
).fetchall()
print(rows)

[('Jansen',)]
[(2, 'Dublin')]

The placeholder style depends on the driver. sqlite3 uses ? and :name; psycopg (PostgreSQL) and the MySQL drivers use %s and %(name)s:

# psycopg (PostgreSQL): %s is a placeholder here, NOT Python string formatting
cur.execute("SELECT name FROM products WHERE category = %s", ("Cables",))

Placeholders stand for values only. Table and column names cannot be parameters; if those must vary, choose them from a fixed list in the code. The full background, including database permissions, is on Database Security.

Transactions in code

Placing an order means two changes: record the order line and lower the stock. Both must happen, or neither (see Transactions and ACID). In Python, a connection used in a with block does exactly that: it commits when the block finishes normally and rolls back when an exception leaves the block.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.executescript("""
    CREATE TABLE products (product_id INTEGER PRIMARY KEY, name TEXT,
                           stock INTEGER NOT NULL CHECK (stock >= 0));
    CREATE TABLE order_lines (order_id INTEGER, product_id INTEGER, quantity INTEGER);
    INSERT INTO products VALUES (5, '27-inch monitor', 7);
""")


def place_order(order_id, product_id, quantity):
    try:
        with conn:  # commits if the block succeeds, rolls back on an exception
            conn.execute(
                "INSERT INTO order_lines VALUES (?, ?, ?)",
                (order_id, product_id, quantity),
            )
            conn.execute(
                "UPDATE products SET stock = stock - ? WHERE product_id = ?",
                (quantity, product_id),
            )
        print(f"order {order_id}: committed")
    except sqlite3.IntegrityError as error:
        print(f"order {order_id}: rolled back ({error})")


place_order(201, 5, 2)
place_order(202, 5, 9)  # only 5 left: the CHECK constraint fails

print(conn.execute("SELECT stock FROM products WHERE product_id = 5").fetchone())
print(conn.execute("SELECT * FROM order_lines ORDER BY order_id").fetchall())
conn.close()

order 201: committed
order 202: rolled back (CHECK constraint failed: stock >= 0)
(5,)
[(201, 5, 2)]

Order 202's INSERT succeeded, but the UPDATE would have made the stock negative. The exception made the with block roll back, so the order line of 202 is gone as well: stock 5, one order line. Without the transaction, the shop would have an order line for monitors it never took out of stock.

Handling database errors

Some errors are normal events that the application should handle: a customer signs up with an email address that already exists, and the UNIQUE constraint refuses it. The driver raises an exception; DB-API drivers use a standard set of exception classes (IntegrityError for constraint violations, OperationalError for problems such as a lost connection or a locked database, ProgrammingError for SQL mistakes).

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE customers (customer_id INTEGER PRIMARY KEY, email TEXT NOT NULL UNIQUE)")
conn.execute("INSERT INTO customers (email) VALUES (?)", ("mei.chen@example.com",))
conn.commit()


def sign_up(email):
    try:
        with conn:
            conn.execute("INSERT INTO customers (email) VALUES (?)", (email,))
        return "Account created."
    except sqlite3.IntegrityError as error:
        # Log the details for the developers; show the user a calm message.
        print("log:", type(error).__name__, error)
        return "An account with this email address already exists."


print(sign_up("ana.silva@example.com"))
print(sign_up("mei.chen@example.com"))
print(conn.execute("SELECT COUNT(*) FROM customers").fetchone()[0])

Account created.
log: IntegrityError UNIQUE constraint failed: customers.email
An account with this email address already exists.
2

Letting the database enforce uniqueness and catching the error is more reliable than first checking "does this email exist?" and then inserting: between the check and the insert, another request could insert the same address (a race condition). Two rules for error handling:

• Catch the specific exception you expect and can handle. Let unexpected errors reach the application's general error handler, which logs them.
• Never show raw database errors to users. They reveal table and column names that help an attacker; log them on the server and show a generic message.

Object-relational mappers

Writing SQL by hand and copying columns into objects becomes repetitive in large applications. Object-relational mapping (ORM) lets you define classes that correspond to tables; the ORM library generates the SQL, sends it with parameters, and turns rows into objects. Well-known ORMs are SQLAlchemy and the Django ORM for Python, Entity Framework (Core) for .NET, Hibernate for Java, and ActiveRecord in Ruby on Rails.

A short SQLAlchemy (version 2) sketch of two shop tables, not run here:

from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship


class Base(DeclarativeBase):
    pass


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


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


engine = create_engine("sqlite:///shop.db")
with Session(engine) as session:
    sanne = session.scalars(select(Customer).where(Customer.first_name == "Sanne")).one()
    for order in sanne.orders:  # the ORM runs a query here to load the orders
        print(order.order_id, order.status)

The benefits: less repetitive code, parameters used automatically, and the same code can work with several databases. The costs: another layer to learn, and the SQL is out of sight. An ORM makes simple things easy and complicated reports awkward; most ORMs let you write raw SQL where that is clearer. Turn on the ORM's SQL logging during development so you see what it actually sends.

The N+1 query problem

The loop over sanne.orders above hides a query. Do that for every customer in a list and you get the N+1 query problem: one query fetches N rows, then the code runs one more query for each of them. Each query is fast, but every one costs a round trip to the server; for 500 customers that is 501 queries where one would do. The example counts the statements with set_trace_callback, which calls a function for every statement sqlite3 sends:

import sqlite3

conn = sqlite3.connect(":memory:")
conn.executescript("""
    CREATE TABLE customers (customer_id INTEGER PRIMARY KEY, first_name TEXT);
    CREATE TABLE orders (order_id INTEGER PRIMARY KEY, customer_id INTEGER, order_date TEXT);
    INSERT INTO customers VALUES (1, 'Sanne'), (2, 'Kwame'), (3, 'Mei'), (4, 'Omar');
    INSERT INTO orders VALUES (101, 1, '2026-06-02'), (102, 2, '2026-06-05'),
                              (103, 1, '2026-06-18'), (104, 3, '2026-07-01');
""")

statements = []
conn.set_trace_callback(statements.append)  # records every SQL statement sent

# N+1: one query for the customers, then one more query per customer
customers = conn.execute("SELECT customer_id, first_name FROM customers ORDER BY customer_id")
for customer_id, name in customers:
    count = conn.execute(
        "SELECT COUNT(*) FROM orders WHERE customer_id = ?", (customer_id,)
    ).fetchone()[0]
    print(name, count)
print("queries:", len(statements))

statements.clear()

# One query: let the database join and count
rows = conn.execute("""
    SELECT c.first_name, COUNT(o.order_id)
    FROM customers AS c
    LEFT JOIN orders AS o ON o.customer_id = c.customer_id
    GROUP BY c.customer_id
    ORDER BY c.customer_id
""").fetchall()
for name, count in rows:
    print(name, count)
print("queries:", len(statements))

Sanne 2
Kwame 1
Mei 1
Omar 0
queries: 5
Sanne 2
Kwame 1
Mei 1
Omar 0
queries: 1

Same answer, five queries against one. The fix is to ask the database for everything at once: a join with GROUP BY, or WHERE customer_id IN (...). ORMs have options for this, called eager loading (in SQLAlchemy, for example, selectinload; in Django, select_related and prefetch_related).

Connection pools

Opening a connection to a database server costs tens of milliseconds and uses memory on the server; many servers handle only a limited number of connections (PostgreSQL's default max_connections is 100). A web application that opened a new connection for every request would be slow and would run out of connections under load. A connection pool keeps a set of open connections: a request borrows one, uses it and returns it, and the next request reuses it.

Pools are built into most frameworks and ORMs (SQLAlchemy's engine is a pool, as are JDBC pools such as HikariCP). Points to know:

• Pool size × number of application instances must stay below the database's connection limit. Ten containers with a pool of 20 each want 200 connections.
• Return connections promptly. Code that forgets to release a connection (a "connection leak") empties the pool until every request waits.
• External poolers such as PgBouncer for PostgreSQL sit between many application instances and the database, and are common with serverless functions, which run as many short-lived instances that together open far too many connections.

Schema migrations as code

The application's code expects certain tables and columns. When a new version needs a new column, the database has to change at the same moment, on every environment: developers' laptops, test, production. A schema migration is a versioned script that changes the schema from one version to the next (and, ideally, back). Migrations live in the repository next to the code, and a tool applies the ones a database has not had yet, recording which ran in a table of its own.

migrations/
├── V1__create_customers_and_products.sql
├── V2__create_orders.sql
├── V3__add_stock_to_products.sql
└── V4__add_index_orders_customer.sql

Common tools are Alembic (for SQLAlchemy), Django's built-in migrations, Entity Framework migrations, Flyway and Liquibase. Because every change goes through a reviewed, tested script, nobody edits the production schema by hand, and every environment ends up identical. How to change a large production table safely is covered in Database Administration in Practice.