Published / updated29 Aug 2026

Databases and REST APIs

Before you start

You need: exceptions (Article 05) and environments (Article 07).

Time: about 50 minutes, plus the practice.

Learning objective

Query a database safely with parameters, and call a REST API handling every status code and failure mode.

Topics

• DB-API basics
• Parameters and injection
• Transactions
• Mapping rows to objects
• SQL Server from Python
• Calling REST APIs
• Status codes and errors
• Retries and timeouts
• Configuration and secrets

DB-API

Python's database interface is standardised, so sqlite3, pyodbc and psycopg share the same shape.

import sqlite3
from pathlib import Path

with sqlite3.connect(Path("school.db")) as connection:
    cursor = connection.cursor()

    cursor.execute("""
        SELECT Id, RollNumber, Name, ClassName
        FROM   Student
        WHERE  SchoolId = ? AND Status <> 1
        ORDER BY ClassName, Name
    """, (school_id,))

    rows = cursor.fetchall()

Method	Returns
fetchone()	One row, or None
fetchall()	Every row as a list
fetchmany(n)	Up to n rows
Iterating the cursor	One row at a time, streaming

fetchall() loads everything into memory. For a large result, iterate the cursor instead:

for row in cursor.execute(sql, params):
    process(row)

sqlite3.connect as a context manager commits or rolls back the transaction — it does not close the connection. Close it explicitly:

from contextlib import closing

with closing(sqlite3.connect(path)) as connection:
    with connection:                      # transaction
        connection.execute(sql, params)

That surprises people: the with most tutorials show leaves the connection open.

Parameters

Never build SQL with string formatting.

# Injection, and it breaks on any apostrophe
cursor.execute(f"SELECT * FROM Student WHERE Name = '{name}'")
cursor.execute("SELECT * FROM Student WHERE Name = '%s'" % name)

A parent named O'Brien breaks the query. A value of '; DROP TABLE Student; -- does considerably worse.

# sqlite3, psycopg — qmark or pyformat
cursor.execute("SELECT * FROM Student WHERE SchoolId = ? AND ClassName = ?",
               (school_id, class_name))

# Named parameters
cursor.execute("SELECT * FROM Student WHERE SchoolId = :school_id",
               {"school_id": school_id})

The parameters go in a tuple as the second argument, not formatted into the string. The driver sends them separately, so the value can never become SQL.

A single parameter needs a trailing comma — (school_id,), not (school_id), which is just the integer.

Identifiers cannot be parameters. A table or column name chosen at run time must come from a whitelist:

SORT_COLUMNS = {"name": "Name", "roll": "RollNumber", "class": "ClassName, Section"}

order_by = SORT_COLUMNS.get(requested_sort, "Name")

cursor.execute(f"SELECT ... FROM Student WHERE SchoolId = ? ORDER BY {order_by}", (school_id,))

The f-string interpolates only one of three fixed strings the code controls — never the caller's input.

# IN with a variable-length list
placeholders = ",".join("?" for _ in student_ids)

cursor.execute(
    f"SELECT * FROM Student WHERE Id IN ({placeholders})",
    tuple(student_ids)
)

The placeholders are generated, so nothing from the caller reaches the SQL text.

Transactions

def record_payment(connection, school_id: int, account_id: int, amount: Decimal) -> int:
    cursor = connection.cursor()

    try:
        cursor.execute("""
            INSERT INTO FeePayment (SchoolId, FeeAccountId, Amount, PaidOn)
            VALUES (?, ?, ?, ?)
        """, (school_id, account_id, str(amount), datetime.now(timezone.utc)))

        payment_id = cursor.lastrowid

        cursor.execute("""
            UPDATE FeeAccount
            SET    PaidAmount = PaidAmount + ?
            WHERE  Id = ? AND SchoolId = ?
        """, (str(amount), account_id, school_id))

        if cursor.rowcount == 0:
            raise ValueError("Fee account not found")

        connection.commit()
        return payment_id

    except Exception:
        connection.rollback()
        raise

Both statements must succeed or neither. Without the transaction, a failure between them records a payment against an unchanged balance — the receipt says paid, the system says owing.

cursor.rowcount after an UPDATE tells you whether anything matched. An update matching nothing is not an error, so without the check the function reports success having done nothing.

SchoolId appears in every WHERE clause, including the update. The tenant filter belongs in the SQL, so no future function can omit it.

Note the Decimal is passed as a string — SQLite has no decimal type, and passing the Decimal directly stores a float.

Mapping rows to objects

connection.row_factory = sqlite3.Row

cursor.execute("SELECT Id, RollNumber, Name FROM Student WHERE Id = ?", (student_id,))
row = cursor.fetchone()

if row is not None:
    print(row["Name"])        # by name, not by index

Index access breaks silently when the SELECT list changes. row[2] is correct until someone adds a column, and then it returns the wrong value with no error — the worst kind of failure.

def map_student(row: sqlite3.Row) -> Student:
    return Student(
        id=row["Id"],
        roll_number=row["RollNumber"],
        name=row["Name"],
        class_name=row["ClassName"],
        marks=row["Marks"] if row["Marks"] is not None else None
    )


def get_students(connection, school_id: int) -> list[Student]:
    cursor = connection.cursor()

    cursor.execute("""
        SELECT Id, RollNumber, Name, ClassName, Marks
        FROM   Student
        WHERE  SchoolId = ? AND Status <> 1
        ORDER BY ClassName, Name, Id
    """, (school_id,))

    return [map_student(row) for row in cursor.fetchall()]

One mapper per entity, used by every query returning it. Mapping inline in each function guarantees one of them eventually forgets a null check.

NULL becomes None automatically — the mapper's job is to preserve that distinction rather than convert it to 0.

The deterministic ORDER BY ending in Id matters as soon as paging is added.

SQL Server

pip install pyodbc

import pyodbc

connection_string = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=localhost;"
    "DATABASE=NexCodingSchool;"
    "UID=app_user;PWD=...;"
    "TrustServerCertificate=yes;"
)

with pyodbc.connect(connection_string) as connection:
    cursor = connection.cursor()

    cursor.execute("""
        SELECT s.Id, s.RollNumber, s.Name
        FROM   dbo.Student AS s
        WHERE  s.SchoolId = ? AND s.Status <> 1
    """, school_id)

    for row in cursor:
        print(row.RollNumber, row.Name)

pyodbc uses ? placeholders and exposes columns as attributes — row.RollNumber.

The ODBC driver must be installed separately from the Python package; pip install pyodbc alone is not enough, and the resulting error names a missing driver.

cursor.execute("{CALL dbo.usp_GetStudentsByClass (?, ?)}", school_id, class_name)

cursor.fast_executemany = True

cursor.executemany(
    "INSERT INTO dbo.Student (SchoolId, Name, RollNumber) VALUES (?, ?, ?)",
    [(school_id, s.name, s.roll_number) for s in students]
)

fast_executemany = True sends the batch in one round trip. Without it, executemany issues one statement per row — the difference between seconds and minutes on ten thousand rows.

SQLAlchemy

pip install sqlalchemy

from sqlalchemy import create_engine, text

engine = create_engine(connection_url, pool_pre_ping=True)

with engine.connect() as connection:
    result = connection.execute(
        text("SELECT Id, Name FROM Student WHERE SchoolId = :school_id"),
        {"school_id": school_id}
    )

    for row in result:
        print(row.Name)

SQLAlchemy Core gives connection pooling and one parameter style across every database. Its ORM adds model mapping and change tracking, like EF Core.

Use SQLAlchemy for anything beyond a small script. Pooling alone justifies it — creating a connection per request does not scale.

Calling REST APIs

pip install requests

import requests

response = requests.get(
    "https://api.nexcoding.in/api/students",
    params={"schoolId": 1, "className": "10th"},
    headers={"Authorization": f"Bearer {token}"},
    timeout=10
)

response.raise_for_status()

data = response.json()

timeout is not optional. Without it a request can hang indefinitely, and a scheduled job that never returns is worse than one that fails.

raise_for_status() raises on 4xx and 5xx. Without it, an HTML error page is parsed as JSON and fails with a confusing message far from the cause.

params builds and encodes the query string — a value of 10th & A works, which string concatenation does not.

response = requests.post(
    "https://api.nexcoding.in/api/students",
    json={"name": "Sneha Patel", "rollNumber": "NCA-2024-0044"},
    headers={"Authorization": f"Bearer {token}"},
    timeout=10
)

json= sets Content-Type: application/json and serialises for you. Using data=json.dumps(...) sends form-encoded content, and an ASP.NET Core [FromBody] endpoint returns 415.

files = {"photo": ("ravi.jpg", open(path, "rb"), "image/jpeg")}
response = requests.post(url, files=files, timeout=30)

With files=, do not set Content-Type — requests sets multipart/form-data with the boundary, and overriding it makes the body unparseable.

Sessions

session = requests.Session()
session.headers.update({"Authorization": f"Bearer {token}"})

response = session.get(f"{base_url}/api/students", timeout=10)

A Session reuses the TCP connection and applies shared headers. For several calls to one API it is measurably faster, and it keeps the token in one place.

Status codes and errors

from requests.exceptions import HTTPError, Timeout, ConnectionError, RequestException


def fetch_students(school_id: int, token: str) -> list[dict]:
    try:
        response = requests.get(
            f"{BASE_URL}/api/students",
            params={"schoolId": school_id},
            headers={"Authorization": f"Bearer {token}"},
            timeout=10
        )
        response.raise_for_status()

    except Timeout as error:
        raise ApiUnavailableError("The API did not respond in time") from error

    except ConnectionError as error:
        raise ApiUnavailableError("Could not reach the API") from error

    except HTTPError as error:
        status = error.response.status_code

        if status == 401:
            raise AuthenticationError("Token missing or expired") from error

        if status == 403:
            raise AuthorizationError("Not permitted") from error

        if status == 404:
            raise NotFoundError(f"No students for school {school_id}") from error

        if status == 400:
            raise ValidationError(error.response.json().get("errors", {})) from error

        raise ApiError(f"Request failed with status {status}") from error

    return response.json()["items"]

Status	Meaning
400	Validation failed — read the errors object
401	Not authenticated — token missing or expired
403	Authenticated, not permitted — wrong role
404	Not found
409	Conflict — duplicate
415	Wrong Content-Type — a bug in your code
429	Rate limited — back off
500+	Server failure

401 versus 403 is the distinction interviewers ask about: 401 means authenticate, 403 means authenticated and refused, so retrying with the same token is pointless.

raise ... from error chains the exceptions, so the traceback shows both the HTTP failure and your domain error.

A parsed response is untrusted data. data["items"][0]["name"] raises KeyError or IndexError when the shape differs. Validate what you depend on:

from pydantic import BaseModel

class StudentResponse(BaseModel):
    publicId: str
    name: str
    rollNumber: str
    className: str

students = [StudentResponse.model_validate(item) for item in data["items"]]

Pydantic validates and converts in one step, and a shape change fails immediately with a message naming the field.

Retries and timeouts

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(
    total=3,
    backoff_factor=1,                          # 1s, 2s, 4s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "PUT", "DELETE"]   # NOT POST
)

session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=retry))

Three rules:

Retry only transient failures — 5xx, 429 and connection errors. Retrying a 400 or 404 fails identically three times and delays the real message.

Back off exponentially. Retrying immediately makes an overloaded server worse.

Never retry a POST without an idempotency key. A timeout does not mean the request failed to arrive — two retries can create three students.

response = requests.get(url, timeout=(5, 30))    # (connect, read)

A separate connect timeout fails fast on an unreachable host while allowing a slow response.

Configuration and secrets

pip install python-dotenv

# .env — NEVER committed
DATABASE_URL=mssql+pyodbc://app_user:...@localhost/NexCodingSchool
API_BASE_URL=https://api.nexcoding.in
API_TOKEN=...

# .env.example — committed, with no real values
DATABASE_URL=
API_BASE_URL=https://api.nexcoding.in
API_TOKEN=

import os
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.environ["DATABASE_URL"]        # KeyError if missing — fail loudly
API_BASE_URL = os.getenv("API_BASE_URL", "https://api.nexcoding.in")

os.environ[...] for anything required. os.getenv returns None, and the failure surfaces later as a confusing connection error rather than a clear startup failure.

Better still, validate everything at startup:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    api_base_url: str = "https://api.nexcoding.in"
    api_token: str
    request_timeout: int = 10

    class Config:
        env_file = ".env"

settings = Settings()      # raises immediately if anything required is missing

A misconfigured deployment then fails at startup with a message naming the setting — far better than failing on the first request that needs it.

# .gitignore
.env
*.db
__pycache__/
.venv/

Never commit .env. A credential in Git history is permanent — deleting it in a later commit does not remove it, and the fix is rotating the credential, not editing history.

git log -p | grep -i "password\|token\|api[_-]key"

If that finds anything, rotate first.

Never log a token, a password or a connection string. Logs are copied to aggregation services and read by more people than the database.

Diagnosing failures

Symptom	Cause
sqlite3.OperationalError: no such table	Wrong database file, or the schema was never created
pyodbc.Error: Data source name not found	ODBC driver not installed
IntegrityError	A constraint was violated — read which
Wrong data after a SELECT change	Index-based row access
KeyError on a response	The API shape differs from what you assumed
JSONDecodeError on a response	An HTML error page — you skipped raise_for_status()
415 from a .NET API	data= instead of json=
A job hangs forever	No timeout

print(Path("school.db").resolve())       # which file
print(cursor.description)                 # the actual column names
print(response.status_code, response.text[:500])

response.text[:500] before parsing shows an HTML error page immediately, where .json() only reports a decode failure.

Errors you will hit

Message	Cause	Fix
sqlite3.OperationalError: no such table	Wrong database file, or table not created	Check the path
sqlite3.ProgrammingError: Incorrect number of bindings	Parameter count mismatch	Match placeholders to values
Data missing after the script ends	Never committed	connection.commit()
A name with an apostrophe breaks the query	String-built SQL	Use parameters
requests.exceptions.ConnectionError	Wrong URL, or no network	Check the URL
JSONDecodeError from an API call	Response was not JSON	Check response.status_code first

requests does not raise on a 404. Check response.status_code or call raise_for_status() before parsing.

Common mistakes

• SQL built with f-strings or %
• A single parameter without the trailing comma
• Row access by index
• No transaction around related writes
• Not checking cursor.rowcount after an update
• The tenant filter missing from one query
• fetchall() on a large result
• A connection per request instead of pooling
• Decimal stored as a float
• No timeout on an HTTP call
• No raise_for_status()
• data=json.dumps(...) instead of json=
• Setting Content-Type with files=
• Retrying a POST
• Retrying 4xx
• Trusting a parsed response without validation
• .env committed
• os.getenv for a required setting
• Logging a token or connection string

Practice

The course exercises are run simple SQL and parse an API response.

1. Create a SQLite database with a Student table and insert five rows with parameters.
2. Query with an f-string containing a name with an apostrophe. Record the error.
3. Query with '; DROP TABLE Student; -- as the name, using an f-string. Then convert to parameters and repeat.
4. Pass a single parameter as (school_id) and record the error. Fix it with the trailing comma.
5. Access rows by index, then add a column to the SELECT list. Confirm the silent wrong data. Switch to sqlite3.Row.
6. Write record_payment with a transaction. Force a failure between the two statements and confirm neither happened.
7. Remove the transaction and repeat. Confirm the inconsistent state.
8. Update a non-existent row and confirm rowcount is 0. Add the check.
9. Build an IN clause with generated placeholders for four ids.
10. Add a run-time sort column from a whitelist. Pass Name; DROP TABLE Student-- and confirm it falls through.
11. Call an API without raise_for_status(), pointed at a 404. Record what .json() does.
12. Call it without timeout against a slow endpoint.
13. POST with data=json.dumps(...) to a .NET API and confirm the 415. Switch to json=.
14. Handle 400, 401, 403, 404 and 500 separately with domain exceptions.
15. Validate a response with a pydantic model. Change a field name on the server and confirm the immediate failure.
16. Add retries for 5xx only. Confirm a 400 is not retried.
17. Move configuration to .env with pydantic-settings. Remove a required value and confirm the startup failure.
18. Run git log -p | grep -i token on your repository.

Exercises 3, 5 and 7 correspond to an injection hole, silent data corruption and an inconsistent balance.

You can now

• Query a database with parameters and transactions
• Commit explicitly and say what happens without it
• Call a REST API and check the status code first
• Handle every failure mode of a network call
• Keep credentials out of the source

Review questions

1. Why can a column name never be a query parameter, and what is the safe alternative?
2. Why must cursor.rowcount be checked after an update?
3. Why is json= required rather than data=json.dumps(...)?
4. Which HTTP failures are safe to retry, and which are not?

Next: FastAPI introduction