Data Analysis with Python (pandas, polars, DuckDB) Module 3 – Loading data: CSV, Excel, JSON, Parquet and databases

Reading from SQL databases safely

Load SQL query results into pandas with read_sql: pass values as parameters, allowlist column names, group in the database and write back with to_sql.

Intermediate
25 minutes
Examples run with Python 3.14.8, pandas 3.0.6, numpy 2.5.3 and sqlalchemy 2.0.48
By MySmartCoPilot

What you will learn

Load query results into pandas with read_sql from SQLite and other databases
Pass values as query parameters and allowlist identifiers instead of formatting SQL strings
Move filtering and aggregation into the database when the data is large
Write DataFrames back with to_sql and choose an if_exists policy

Before you start

Reading messy CSV files with pandas

What you’ll learn · Before you start

What you will learn

Load query results into pandas with read_sql from SQLite and other databases
Pass values as query parameters and allowlist identifiers instead of formatting SQL strings
Move filtering and aggregation into the database when the data is large
Write DataFrames back with to_sql and choose an if_exists policy

Before you start

Reading messy CSV files with pandas

On this page

pd.read_sql(query, connection) runs a SQL query and returns its result as a DataFrame. With Python’s built-in sqlite3 module it needs nothing else; other databases connect through SQLAlchemy or an ADBC driver. Doing it safely comes down to three habits. Values such as a department name always travel as parameters, never pasted into the SQL text, because pasted text can change what the query does. Names of columns and tables cannot be parameters, so they are checked against an allowlist. And summaries are computed in the database with WHERE and GROUP BY, so only the small answer travels to Python.

The other direction is DataFrame.to_sql, which writes a DataFrame into a table. Its if_exists option decides what happens to a table that is already there, and one of its choices quietly removes the table’s rules. This lesson works on one SQLite database of exam results, with every query run and recorded.

A results database to work with

The examples build the same SQLite database: 20,000 students in four departments and 100,000 exam results, about 1 % of them missing. The helper, imported by every example:

make_college.py, imported by the examples Python · make_college.py

"""Builds college.db, a SQLite database of 20,000 students and 100,000 exam results (seeded)."""
import sqlite3

import numpy as np

FIRST = ["Asha", "Ben", "Chen", "Divya", "Emre", "Farah", "Gopal", "Hana", "Ivan", "Jomo", "Kavya", "Liam"]
DEPARTMENTS = ["CSE", "ECE", "MECH", "CIVIL"]
SUBJECTS = ["Maths", "Physics", "Chemistry", "English", "Programming"]


def build(path="college.db", students=20_000, seed=5):
    rng = np.random.default_rng(seed)
    con = sqlite3.connect(path)
    con.executescript("""
        DROP TABLE IF EXISTS results;
        DROP TABLE IF EXISTS students;
        CREATE TABLE students (
            roll_no TEXT PRIMARY KEY, name TEXT NOT NULL, department TEXT NOT NULL, year INTEGER NOT NULL);
        CREATE TABLE results (
            roll_no TEXT NOT NULL REFERENCES students (roll_no), subject TEXT NOT NULL,
            marks INTEGER, exam_date TEXT NOT NULL);
    """)
    people = [
        (f"{i:05d}", f"{FIRST[int(rng.integers(0, len(FIRST)))]} {i}", DEPARTMENTS[i % 4], 1 + (i // 4) % 4)
        for i in range(1, students + 1)
    ]
    con.executemany("INSERT INTO students VALUES (?, ?, ?, ?)", people)
    marks = rng.integers(20, 100, (students, len(SUBJECTS)))
    absent = rng.random((students, len(SUBJECTS))) < 0.01  # about 1 % of results are missing
    days = rng.integers(11, 20, (students, len(SUBJECTS)))
    con.executemany(
        "INSERT INTO results VALUES (?, ?, ?, ?)",
        [
            (roll, subject, None if absent[i, j] else int(marks[i, j]), f"2026-05-{days[i, j]}")
            for i, (roll, *_) in enumerate(people)
            for j, subject in enumerate(SUBJECTS)
        ],
    )
    con.commit()
    con.close()
    return path

Values travel as parameters

Write a ? wherever a value belongs, and pass the values separately with params, in the same order. The driver hands them to the database apart from the SQL text, so a value can only ever be a value. Longer queries read better with named placeholders such as :dept, filled from a dictionary:

Placeholders in order and by name Python · param_query.py

import sqlite3

import pandas as pd

from make_college import build

con = sqlite3.connect(build())

# Values travel separately from the SQL text: ? placeholders, filled in order from params.
second_years = pd.read_sql(
    "SELECT roll_no, name FROM students WHERE department = ? AND year = ? ORDER BY roll_no",
    con,
    params=("CSE", 2),
)
print(f"{len(second_years)} students", end="\n\n")
print(second_years.head(3), end="\n\n")

# Named placeholders read better in longer queries: :name, filled from a dictionary.
counts = pd.read_sql(
    "SELECT year, count(*) AS students FROM students WHERE department = :dept GROUP BY year ORDER BY year",
    con,
    params={"dept": "ECE"},
)
print(counts)
con.close()

Output

1250 students

  roll_no      name
0   00004    Jomo 4
1   00020   Liam 20
2   00036  Farah 36

   year  students
0     1      1250
1     2      1250
2     3      1250
3     4      1250

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 param_query.py

Placeholder styles differ between drivers. sqlite3 accepts ? and :name; PostgreSQL’s psycopg uses %s and %(name)s; PEP 249, the specification of Python’s database interface, lists the five styles a driver may choose from. SQLAlchemy’s text() always uses :name, whatever the database.

Why a formatted query is dangerous

It is tempting to build the query with an f-string: f"... WHERE department = '{department}'". If department comes from a form, a file or an API, whoever wrote that value also wrote part of your SQL:

The same input, formatted and as a parameter Python · injection.py

import sqlite3

import pandas as pd

from make_college import build

con = sqlite3.connect(build())

# What a search box received: a department name that closes the quote and adds a condition.
typed = "CSE' OR '1'='1"

unsafe_sql = f"SELECT count(*) AS students FROM students WHERE department = '{typed}'"
print(unsafe_sql)
print(pd.read_sql(unsafe_sql, con), end="\n\n")  # every student in the college

safe = pd.read_sql("SELECT count(*) AS students FROM students WHERE department = ?", con, params=(typed,))
print(safe)  # no department has that strange name
con.close()

Output

SELECT count(*) AS students FROM students WHERE department = 'CSE' OR '1'='1'
   students
0     20000

   students
0         0

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 injection.py

The typed value closed the quote and added OR '1'='1', a condition that is always true, so the formatted query counted every student in the college. As a parameter, the same text is only a strange department name that nobody has. This is SQL injection. OWASP’s cheat sheets put parameterised queries first among the defences, and a value pasted into SQL can do far worse than count rows: it can read other tables or change data, depending on what the connection is allowed to do. pandas’ own documentation notes that it does not sanitise SQL statements.

Names cannot be parameters: use an allowlist

Placeholders stand for values only. A column name in ORDER BY, a table name or a sort direction has to be part of the SQL text, and a placeholder in its place does not fail; it silently does something else:

A placeholder in ORDER BY, and an allowlist Python · allowlist.py

import sqlite3

import pandas as pd

from make_college import build

con = sqlite3.connect(build())

# A placeholder cannot stand for a column name: here SQLite sorts by the constant text 'total'.
ignored = pd.read_sql(
    "SELECT s.name, sum(r.marks) AS total FROM students s JOIN results r USING (roll_no) "
    "WHERE s.department = 'MECH' AND s.year = 4 GROUP BY s.roll_no ORDER BY ? DESC LIMIT 3",
    con,
    params=("total",),
)
print(ignored, end="\n\n")

SORTABLE = {"total": "total DESC", "name": "s.name", "roll_no": "s.roll_no"}


def ranking(department, year, sort_by="total"):
    if sort_by not in SORTABLE:  # identifiers are checked against an allowlist, never pasted in
        raise ValueError(f"cannot sort by {sort_by!r}; choose one of {sorted(SORTABLE)}")
    sql = (
        "SELECT s.roll_no, s.name, sum(r.marks) AS total FROM students s JOIN results r USING (roll_no) "
        f"WHERE s.department = ? AND s.year = ? GROUP BY s.roll_no ORDER BY {SORTABLE[sort_by]} LIMIT 3"
    )
    return pd.read_sql(sql, con, params=(department, year))


print(ranking("MECH", 4), end="\n\n")
try:
    ranking("MECH", 4, sort_by="total; DROP TABLE students")
except ValueError as err:
    print("ValueError:", err)
con.close()

Output

          name  total
0  Farah 19998    273
1   Asha 19982    331
2  Kavya 19966    288

  roll_no         name  total
0   01070   Farah 1070    451
1   14206  Divya 14206    432
2   01518    Hana 1518    431

ValueError: cannot sort by 'total; DROP TABLE students'; choose one of ['name', 'roll_no', 'total']

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 allowlist.py

ORDER BY ? sorted by the constant text 'total', which is the same for every row, so the “top three” are three arbitrary students. The fix is an allowlist: a dictionary from the names a caller may ask for to the exact SQL fragments the code wrote itself. Anything else raises an error before any SQL exists, which is the input-validation rule OWASP recommends: accept known good values rather than trying to reject known bad ones.

Let the database do the summarising

A database can filter, join and aggregate where the data lives. Asking for the summary sends a handful of rows to Python; asking for every row and summarising in pandas sends all of them:

Pushing a summary into the database, or pulling every row into pandas

Text description of the diagram

Both paths start from college.db, which holds 100,000 exam results of 20,000 students, and both compute the average mark of each department.

Push down: the query itself groups and averages (SELECT department, avg(marks) ... GROUP BY department). The database sends 4 rows to Python, 243 bytes in the lesson's run.
Pull up: the query selects every result with its department and no GROUP BY. The database sends 100,000 rows, about 2 MB in memory, and pandas then groups them and computes the means.

The answers are identical; what differs is how much data has to travel and be held in memory.

The same averages, grouped in SQL or in pandas Python · push_down.py

import sqlite3

import pandas as pd

from make_college import build

con = sqlite3.connect(build())

in_database = pd.read_sql(
    "SELECT s.department, round(avg(r.marks), 2) AS avg_marks, count(r.marks) AS sat "
    "FROM results r JOIN students s USING (roll_no) GROUP BY s.department ORDER BY s.department",
    con,
)
every_row = pd.read_sql("SELECT s.department, r.marks FROM results r JOIN students s USING (roll_no)", con)
in_pandas = (
    every_row.groupby("department", as_index=False)
    .agg(avg_marks=("marks", "mean"), sat=("marks", "count"))
    .round({"avg_marks": 2})
)

print(in_database, end="\n\n")
print("same answer:", in_database.equals(in_pandas))
print(f"rows sent to Python: {len(in_database)} against {len(every_row):,}")
print(f"memory of what was sent: {in_database.memory_usage(deep=True).sum():,} bytes "
      f"against {every_row.memory_usage(deep=True).sum():,} bytes")
con.close()

Output

  department  avg_marks    sat
0      CIVIL      59.46  24761
1        CSE      59.62  24741
2        ECE      59.22  24756
3       MECH      59.61  24766

same answer: True
rows sent to Python: 4 against 100,000
memory of what was sent: 243 bytes against 1,975,132 bytes

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 push_down.py

The answers are identical, but one query sent 4 rows and the other 100,000. Here the database runs inside the same Python process, so moving rows is cheap, and the timings are close:

Timing both, with SQLite in the same process Python · push_down_timing.py

import sqlite3
import time

import pandas as pd

from make_college import build

con = sqlite3.connect(build())


def best_of_three(work):
    best = float("inf")
    for _ in range(3):
        start = time.perf_counter()
        work()
        best = min(best, time.perf_counter() - start)
    return best * 1000


pushed = best_of_three(lambda: pd.read_sql(
    "SELECT s.department, avg(r.marks) FROM results r JOIN students s USING (roll_no) GROUP BY s.department", con))
pulled = best_of_three(lambda: pd.read_sql(
    "SELECT s.department, r.marks FROM results r JOIN students s USING (roll_no)", con).groupby("department")["marks"].mean())
print(f"grouped in the database:     {pushed:6.1f} ms")
print(f"pulled and grouped in pandas: {pulled:6.1f} ms")
con.close()

Output

grouped in the database:       56.1 ms
pulled and grouped in pandas:  271.2 ms

This output changes from run to run: the times differ on every run and on every computer

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 push_down_timing.py

With a database server, every pulled row also crosses the network and must fit in Python’s memory, so the cost of pulling grows with the table while the grouped answer stays at four rows. Filter with WHERE, aggregate with GROUP BY and join in SQL, then bring the result into pandas for the analysis SQL is awkward at. The SQL track’s lesson on GROUP BY covers the grouping itself.

The types that come back

SQLite stores each value as NULL, an integer, a real number, text or a blob; it has no separate date or boolean type. So dates arrive as text, and an integer column with a missing value arrives as float64, as in a CSV:

Default types, and the options that fix them Python · types_from_sql.py

import sqlite3

import pandas as pd

from make_college import build

con = sqlite3.connect(build())
query = "SELECT roll_no, subject, marks, exam_date FROM results LIMIT 1000"

plain = pd.read_sql(query, con)
typed = pd.read_sql(query, con, parse_dates=["exam_date"], dtype_backend="numpy_nullable")
print(pd.DataFrame({"plain": plain.dtypes.astype(str), "parse_dates + numpy_nullable": typed.dtypes.astype(str)}))
print(f"\nmissing marks: {plain['marks'].isna().sum()}")
con.close()

Output

             plain parse_dates + numpy_nullable
roll_no        str                       string
subject        str                       string
marks      float64                        Int64
exam_date      str               datetime64[us]

missing marks: 13

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 types_from_sql.py

parse_dates=["exam_date"] converts the text dates, and dtype_backend="numpy_nullable" gives pandas’ nullable types, so marks stays an integer column with missing values (Int64). Text columns then use the string type, which marks missing values with pd.NA. Other databases do have date and boolean types, and their drivers return them as such.

Other databases: SQLAlchemy and ADBC

For PostgreSQL, MySQL, SQL Server and most others, pandas works with an SQLAlchemy engine, created from a connection URL. The same database through SQLAlchemy, with a named parameter and a read in chunks:

SQLAlchemy, text() and chunksize Python · sqlalchemy_engine.py

import pandas as pd
from sqlalchemy import create_engine, text

from make_college import build

# The same database through SQLAlchemy. For another database only the URL changes, for example
# "postgresql+psycopg://analyst@db.example.internal/college", with the password kept out of the code.
engine = create_engine(f"sqlite:///{build()}")

query = text("SELECT department, count(*) AS students FROM students WHERE year = :year GROUP BY department ORDER BY department")
with engine.connect() as connection:
    print(pd.read_sql(query, connection, params={"year": 1}))

    total = 0
    for chunk in pd.read_sql(text("SELECT marks FROM results"), connection, chunksize=40_000):
        total += len(chunk)
    print(f"read in chunks: {total:,} results")
engine.dispose()

Output

  department  students
0      CIVIL      1250
1        CSE      1250
2        ECE      1250
3       MECH      1250
read in chunks: 100,000 results

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 sqlalchemy_engine.py

create_engine(url) takes a URL such as postgresql+psycopg://user@host/database; for a different database only the URL changes.
text() marks a string as SQL with :name parameters, and params fills them.
chunksize=40_000 returns an iterator of DataFrames, as in read_csv, for results too big to hold at once.
ADBC drivers (Arrow Database Connectivity) are a newer alternative for PostgreSQL, SQLite and others: pandas accepts their connections directly and moves data in Arrow’s columnar format, with native types.

Security

Keep passwords out of code and notebooks. Read them from an environment variable or a secrets store when the engine is created, and use a database account that can only read the tables the analysis needs, so that even a mistake in a query cannot change data.

Writing back with to_sql

DataFrame.to_sql(name, connection) creates a table from the DataFrame’s columns and inserts its rows. pandas chooses the column types, and the mapping is approximate:

Six kinds of column written to SQLite and read back Python · round_trip.py

import sqlite3

import pandas as pd

con = sqlite3.connect(":memory:")
fees = pd.DataFrame({
    "roll_no": ["00042", "00107"],
    "paid": [True, False],
    "paid_on": pd.to_datetime(["2026-04-06", "2026-04-07"]),
    "plan": pd.Categorical(["yearly", "termly"]),
    "instalments": pd.array([1, None], dtype="Int64"),
    "amount": [1200.0, 450.5],
})
fees.to_sql("fees", con, index=False)
print(con.execute("SELECT sql FROM sqlite_master WHERE name = 'fees'").fetchone()[0], end="\n\n")

back = pd.read_sql("SELECT * FROM fees", con)
print(pd.DataFrame({"written": fees.dtypes.astype(str), "read back": back.dtypes.astype(str)}))
con.close()

Output

CREATE TABLE "fees" (
"roll_no" TEXT,
  "paid" INTEGER,
  "paid_on" TIMESTAMP,
  "plan" TEXT,
  "instalments" INTEGER,
  "amount" REAL
)

                    written read back
roll_no                 str       str
paid                   bool     int64
paid_on      datetime64[us]       str
plan               category       str
instalments           Int64   float64
amount              float64   float64

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 round_trip.py

The booleans came back as 0 and 1, the dates as text, the category as text and the nullable integers as float64. index=False keeps the meaningless 0, 1, 2 index out of the table. For a table that matters, create it yourself with proper types, keys and checks, and let to_sql only add rows to it. if_exists decides what happens when the table exists:

fail, append, delete_rows and replace Python · if_exists.py

import sqlite3

import pandas as pd

con = sqlite3.connect(":memory:")
TABLE = "SELECT sql FROM sqlite_master WHERE name = 'fees'"
con.execute("CREATE TABLE fees (roll_no TEXT PRIMARY KEY, amount REAL NOT NULL CHECK (amount > 0))")
first = pd.DataFrame({"roll_no": ["00042", "00107"], "amount": [1200.0, 300.0]})

try:
    first.to_sql("fees", con, index=False)  # if_exists="fail" is the default
except ValueError as err:
    print("fail:       ", err)

first.to_sql("fees", con, if_exists="append", index=False)
try:
    pd.DataFrame({"roll_no": ["00042"], "amount": [50.0]}).to_sql("fees", con, if_exists="append", index=False)
except pd.errors.DatabaseError as err:
    print("append:     ", type(err.__cause__).__name__, err.__cause__)

# delete_rows empties the table and inserts, keeping its definition and constraints.
pd.DataFrame({"roll_no": ["00042"], "amount": [1250.0]}).to_sql("fees", con, if_exists="delete_rows", index=False)
print("delete_rows:", con.execute(TABLE).fetchone()[0])
print(pd.read_sql("SELECT * FROM fees", con), end="\n\n")

# replace drops the table and creates a new one from the DataFrame's columns, without any constraint.
bad = pd.DataFrame({"roll_no": ["00042", "00042"], "amount": [1200.0, -5.0]})
bad.to_sql("fees", con, if_exists="replace", index=False)
print("replace:    ", " ".join(con.execute(TABLE).fetchone()[0].split()))
print(pd.read_sql("SELECT * FROM fees", con))
con.close()

Output

fail:        Table 'fees' already exists.
append:      IntegrityError UNIQUE constraint failed: fees.roll_no
delete_rows: CREATE TABLE fees (roll_no TEXT PRIMARY KEY, amount REAL NOT NULL CHECK (amount > 0))
  roll_no  amount
0   00042  1250.0

replace:     CREATE TABLE "fees" ( "roll_no" TEXT, "amount" REAL )
  roll_no  amount
0   00042  1200.0
1   00042    -5.0

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 if_exists.py

"fail", the default, raises ValueError, so nothing is overwritten by accident.
"append" inserts rows into the existing table, and the table’s own rules still apply: the duplicate roll number was refused with an IntegrityError.
"delete_rows", new in pandas 3, empties the table and inserts the new rows, keeping its definition and constraints.
"replace" drops the table and creates a new one from the DataFrame. The primary key, NOT NULL and the check that amounts are positive are gone, and the duplicate roll number and the negative amount went straight in.

As a rule: append for regular loads into a table you defined, delete_rows for full refreshes of such a table, and replace only for scratch tables whose rules nobody relies on.

Your turn: the top students, safely

The starting point below works for ordinary input, but it pastes every value into the SQL text. Rewrite top_students(db_path, department, year, sort_by="total", limit=3) so that the department, the year and the limit are parameters, and sort_by accepts only "total" (highest first), "name" (A to Z) and "roll_no", raising ValueError for anything else:

exercise/top_students.py (your starting point) Python · exercise/top_students.py

import sqlite3

import pandas as pd


def top_students(db_path, department, year, sort_by="total", limit=3):
    """The top students of one department and year: roll_no, name and total marks."""
    sql = f"""
        SELECT s.roll_no, s.name, sum(r.marks) AS total
        FROM students s JOIN results r USING (roll_no)
        WHERE s.department = '{department}' AND s.year = {year}
        GROUP BY s.roll_no
        ORDER BY {sort_by} DESC
        LIMIT {limit}
    """  # every value is pasted into the SQL text: rewrite this
    con = sqlite3.connect(db_path)
    try:
        return pd.read_sql(sql, con)
    finally:
        con.close()

exercise/small_college.py (builds the test database) Python · exercise/small_college.py

"""Builds a small results database for the exercise: 400 students, 5 subjects each (seeded)."""
import sqlite3

import numpy as np


def build(path="small_college.db", seed=9):
    rng = np.random.default_rng(seed)
    con = sqlite3.connect(path)
    con.executescript("""
        DROP TABLE IF EXISTS results;
        DROP TABLE IF EXISTS students;
        CREATE TABLE students (roll_no TEXT PRIMARY KEY, name TEXT NOT NULL, department TEXT NOT NULL, year INTEGER NOT NULL);
        CREATE TABLE results (roll_no TEXT NOT NULL, subject TEXT NOT NULL, marks INTEGER);
    """)
    names = ["Anil", "Bea", "Chidi", "Divya", "Emre", "Fatima", "Gopal", "Hana", "Ines", "Jomo"]
    people = [(f"{i:04d}", f"{names[i % 10]} {i}", ["CSE", "ECE"][i % 2], 1 + (i // 2) % 4) for i in range(1, 401)]
    con.executemany("INSERT INTO students VALUES (?, ?, ?, ?)", people)
    con.executemany(
        "INSERT INTO results VALUES (?, ?, ?)",
        [(roll, subject, int(rng.integers(30, 100))) for roll, *_ in people for subject in ["Maths", "Physics", "Chemistry", "English", "Programming"]],
    )
    con.commit()
    con.close()
    return path

exercise/check_top.py, run on the starting point Python · exercise/check_top.py

import sqlite3

import pandas as pd

from small_college import build
from top_students import top_students

db = build()
con = sqlite3.connect(db)
expected = pd.read_sql(
    "SELECT s.roll_no, s.name, sum(r.marks) AS total FROM students s JOIN results r USING (roll_no) "
    "WHERE s.department = 'CSE' AND s.year = 2 GROUP BY s.roll_no ORDER BY total DESC, s.roll_no LIMIT 3",
    con,
)


def check(name, test):
    try:
        ok = bool(test())
    except Exception:  # an error where none was expected, for example
        ok = False
    print("PASS" if ok else "FAIL", name)
    return ok


def raises_value_error(work):
    try:
        work()
    except ValueError:
        return True
    except Exception:
        return False
    return False


results = [
    check("columns roll_no, name, total", lambda: list(top_students(db, "CSE", 2).columns) == ["roll_no", "name", "total"]),
    check("the top 3 of CSE year 2 by total", lambda: top_students(db, "CSE", 2)["total"].tolist() == expected["total"].tolist()),
    check("limit=5 gives 5 rows", lambda: len(top_students(db, "CSE", 2, limit=5)) == 5),
    check('sort_by="name" sorts names A to Z', lambda: (lambda n: n == sorted(n))(top_students(db, "ECE", 1, sort_by="name")["name"].tolist())),
    check("a crafted department name matches nobody", lambda: len(top_students(db, "CSE' OR '1'='1", 2)) == 0),
    check("an unknown sort column raises ValueError", lambda: raises_value_error(lambda: top_students(db, "CSE", 2, sort_by="total; DROP TABLE students"))),
    check("the students table is still there", lambda: con.execute("SELECT count(*) FROM students").fetchone()[0] == 400),
]
print(f"{sum(results)} of {len(results)} checks pass")
con.close()

Output

PASS columns roll_no, name, total
PASS the top 3 of CSE year 2 by total
PASS limit=5 gives 5 rows
FAIL sort_by="name" sorts names A to Z
FAIL a crafted department name matches nobody
FAIL an unknown sort column raises ValueError
PASS the students table is still there
4 of 7 checks pass

Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 check_top.py

The starting point passes the checks of ordinary use and fails the three that matter: names sort the wrong way, the crafted department name matches students, and the unknown sort column reaches the database. Keep the three files in one place and rewrite top_students.py only; python check_top.py should finish with 7 of 7 checks pass. Hint: LIMIT ? takes a parameter too, and a dictionary from each allowed sort_by to its ORDER BY text gives every choice its own direction.

Interview questions

Fresher · What does the params argument of pd.read_sql do, and why use it instead of an f-string?

params passes values to the database separately from the SQL text. The query contains placeholders, such as ? for sqlite3 or :name with SQLAlchemy’s text(), and the driver binds each value to its placeholder, so a value can only ever be data. With an f-string, the value becomes part of the SQL itself: an input such as CSE' OR '1'='1 closes the quote and adds a condition, which is SQL injection, and the same mistake can break the query on an ordinary name with an apostrophe. Parameters also handle quoting and types for you. They cannot replace column or table names, though; those must come from an allowlist that your own code controls.

Key takeaways

pd.read_sql takes a sqlite3 connection, an SQLAlchemy engine or connection, or an ADBC connection, and returns a DataFrame.
Pass every value through params with placeholders (? or :name in sqlite3); the style depends on the driver.
An f-string query lets input rewrite the SQL: the crafted department name counted all 20,000 students.
Column and table names cannot be parameters (ORDER BY ? silently sorted by a constant); map allowed names to SQL you wrote yourself.
Filter, join and aggregate in SQL: 4 rows instead of 100,000 reached Python for the same answer.
From SQLite, dates arrive as text and integers with missing values as float64; use parse_dates and dtype_backend="numpy_nullable".
to_sql maps types approximately; create important tables yourself and write with append or delete_rows. replace drops the table’s keys and checks.

References

pandas.read_sql (pandas API reference) (pandas development team (NumFOCUS))
pandas.DataFrame.to_sql (pandas API reference) (pandas development team (NumFOCUS))
IO tools: SQL queries (pandas user guide) (pandas development team (NumFOCUS))
sqlite3: How to use placeholders to bind values in SQL queries (Python Software Foundation)
PEP 249: Python Database API Specification v2.0 (paramstyle) (Python Software Foundation)
Datatypes In SQLite (SQLite)
SQLAlchemy 2.0: Engine Configuration (SQLAlchemy project)
Query Parameterization Cheat Sheet (OWASP Foundation)
SQL Injection Prevention Cheat Sheet (OWASP Foundation)
What's new in 3.0.0 (pandas) (pandas development team (NumFOCUS))
Arrow Database Connectivity (ADBC) (The Apache Software Foundation)

Report a problem with this lesson

In this module: Loading data: CSV, Excel, JSON, Parquet and databases

Reading messy CSV files with pandas Free: 35 minutes
Excel workbooks: sheets, headers, dates, writing Free: 30 minutes
JSON, APIs and nested records Free: 30 minutes
Parquet, Arrow and choosing a file format Free: 30 minutes
Reading from SQL databases safely Free: 25 minutes

Your country

Reading from SQL databases safely

What you will learn

Before you start

What you will learn

Before you start

A results database to work with

Values travel as parameters

Why a formatted query is dangerous

Names cannot be parameters: use an allowlist

Let the database do the summarising

The types that come back

Other databases: SQLAlchemy and ADBC

Writing back with to_sql

Your turn: the top students, safely

Check yourself

Interview questions

Key takeaways

References

Related tools