pandas and SQLAlchemy: DataFrames In and Out of SQL

Most data science work that starts in a database ends in a DataFrame: you query the rows you need, then explore, clean, chart or train on them in pandas. Going the other way is just as common: model predictions, cleaned datasets and evaluation scores are written back into tables so dashboards and other services can use them. pandas does both in one line each, and SQLAlchemy is the library that makes those lines work the same on SQLite, MySQL and PostgreSQL.

This page builds on Python and Databases: Connections, Parameters and Transactions. Everything here still runs on DB-API connections underneath; you are adding two layers on top.

What you will learn

• pd.read_sql with parameters, dates and an index, and reading big results in chunks.
• DataFrame.to_sql: if_exists, index, dtype, chunksize and method="multi", and what it does to your schema.
• SQLAlchemy engines and connection URLs built from environment variables, and its connection pool.
• SQLAlchemy Core (text(), select()) and a first look at the ORM (models, Session).
• When to use raw SQL, Core, the ORM or pandas.

From a query to a DataFrame: read_sql

For SQLite, pandas accepts a plain sqlite3 connection. Parameters work exactly as on the previous page:

import sqlite3
import pandas as pd

con = sqlite3.connect("shop.db")

orders = pd.read_sql(
    "SELECT order_id, customer_id, order_date, status FROM orders WHERE status = ? ORDER BY order_id",
    con,
    params=("delivered",),
    index_col="order_id",            # use this column as the row labels
    parse_dates=["order_date"],      # text '2026-01-12' -> a real datetime
)
print(orders.head(3))
print(orders.dtypes)
print(len(orders), "delivered orders")

          customer_id order_date     status
order_id                                   
1                   1 2026-01-12  delivered
2                   2 2026-01-20  delivered
3                   1 2026-02-03  delivered
customer_id             int64
order_date     datetime64[us]
status                    str
dtype: object
11 delivered orders

SQLite stores dates as text, so without parse_dates the column would be str (pandas 3's string dtype) and you could not do date arithmetic on it. Always check dtypes after loading: it is the cheapest bug detector there is.

The healthy split of work is: SQL filters, joins and aggregates (it runs where the data is), pandas does what comes after (reshaping, charts, features, models):

sql = """
    SELECT c.name AS category, SUM(oi.quantity * oi.unit_price) AS revenue
    FROM order_items oi
    JOIN orders o     ON o.order_id = oi.order_id
    JOIN products p   ON p.product_id = oi.product_id
    JOIN categories c ON c.category_id = p.category_id
    WHERE o.status <> 'cancelled'
    GROUP BY c.name
    ORDER BY revenue DESC
"""
by_category = pd.read_sql(sql, con)
by_category["share_pct"] = (100 * by_category["revenue"] / by_category["revenue"].sum()).round(1)
print(by_category)

      category  revenue  share_pct
0      Courses    36000       42.3
1  Electronics    31550       37.1
2  Accessories     8900       10.5
3        Books     8600       10.1

From a DataFrame to a table: to_sql

Suppose a churn model has scored every customer. Writing the scores back takes one call, which creates the table if needed and returns the number of rows written:

scores = pd.DataFrame({
    "customer_id": [1, 2, 3, 4, 5, 6, 7, 8],
    "churn_score": [0.12, 0.35, 0.08, 0.91, 0.47, 0.22, 0.64, 0.88],
    "model": "churn-v1",
})
print(scores.to_sql("churn_scores", con, index=False))
print(con.execute("SELECT sql FROM sqlite_master WHERE name = 'churn_scores'").fetchone()[0])

try:
    scores.to_sql("churn_scores", con, index=False)          # if_exists="fail" is the default
except ValueError as exc:
    print("ValueError:", exc)

8
CREATE TABLE "churn_scores" (
"customer_id" INTEGER,
  "churn_score" REAL,
  "model" TEXT
)
ValueError: Table 'churn_scores' already exists.

Look at the table pandas made: no primary key, no NOT NULL, no constraints at all, and types guessed from the DataFrame. That is fine for a scratch table and wrong for anything other code depends on. The dtype argument overrides the guessed types, as a dict from column name to type: SQLAlchemy types such as {"model": String(20)} with an engine, or type names such as {"model": "TEXT"} with a plain sqlite3 connection. It still adds no keys or constraints. if_exists decides what happens when the table is already there:

The professional pattern is: create the table yourself with proper keys and constraints (DDL from the design pages), then append. Now the database protects the data, for example against scoring the same customer twice. pandas wraps the driver's exception in pd.errors.DatabaseError; the original one is in __cause__:

con.execute("DROP TABLE churn_scores")
con.execute("""
    CREATE TABLE churn_scores (
        customer_id INTEGER PRIMARY KEY,
        churn_score REAL NOT NULL CHECK (churn_score BETWEEN 0 AND 1),
        model       TEXT NOT NULL,
        FOREIGN KEY (customer_id) REFERENCES customers (customer_id)
    )""")
print(scores.to_sql("churn_scores", con, if_exists="append", index=False))
try:
    scores.head(2).to_sql("churn_scores", con, if_exists="append", index=False)
except pd.errors.DatabaseError as exc:                 # pandas wraps the driver's error
    print(type(exc.__cause__).__name__, "-", exc.__cause__)
print(pd.read_sql("SELECT * FROM churn_scores WHERE churn_score > 0.6 ORDER BY customer_id", con))

8
IntegrityError - UNIQUE constraint failed: churn_scores.customer_id
   customer_id  churn_score     model
0            4         0.91  churn-v1
1            7         0.64  churn-v1
2            8         0.88  churn-v1

For many rows, two more options matter. chunksize writes in batches of that many rows (each batch is one executemany), and method="multi" puts many rows in one INSERT … VALUES (…), (…), … statement, which is usually much faster over a network to MySQL or PostgreSQL (measure it: with some drivers it is not). SQLite limits a statement to 32,766 parameters, so with "multi" keep chunksize × columns below that.

import numpy as np

rng = np.random.default_rng(7)
events = pd.DataFrame({
    "event_id": np.arange(1, 20_001),
    "customer_id": rng.integers(1, 9, 20_000),
    "kind": rng.choice(["view", "search", "add_to_cart", "chat"], 20_000),
    "seconds": rng.integers(1, 600, 20_000),
})
written = events.to_sql("events", con, index=False, chunksize=5_000, method="multi")
print(written, "rows written")

20000 rows written

Reading in chunks

With chunksize, read_sql returns an iterator of DataFrames instead of one big one, so you can process a table that does not fit in memory and keep only a small running result. With sqlite3 the rows really are fetched chunk by chunk. With the PostgreSQL and MySQL drivers the default cursor downloads the whole result first and pandas only splits it afterwards; to stream for real, pass a connection opened with engine.connect().execution_options(stream_results=True), which uses a server-side cursor.

totals = None
for i, chunk in enumerate(pd.read_sql("SELECT kind, seconds FROM events ORDER BY event_id", con, chunksize=6_000)):
    part = chunk.groupby("kind")["seconds"].agg(["count", "sum"])
    totals = part if totals is None else totals.add(part, fill_value=0)
    print(f"chunk {i}: {len(chunk)} rows")
print(totals.astype(int).sort_index())

chunk 0: 6000 rows
chunk 1: 6000 rows
chunk 2: 6000 rows
chunk 3: 2000 rows
             count      sum
kind                       
add_to_cart   5081  1531748
chat          5020  1521329
search        4945  1496979
view          4954  1492420

Only partial sums cross chunk boundaries. A mean, for example, has to be built from sums and counts at the end; averaging the chunk means would be wrong when chunks differ in size.

SQLAlchemy: engines and URLs

For any database except SQLite, pandas wants a SQLAlchemy engine (give it a raw PyMySQL or psycopg2 connection and it still runs, but warns that only SQLAlchemy, ADBC and sqlite3 connections are supported). An engine knows how to reach one database, keeps a pool of connections, and translates between your code and that database's dialect. It is described by a URL:

dialect+driver://username:password@host:port/database

sqlite:///shop.db                               (a file next to the script)
sqlite://                                       (in memory)
mysql+pymysql://app:secret@127.0.0.1:3306/shop
postgresql+psycopg2://app:secret@127.0.0.1:5432/shop

Do not glue a URL together with an f-string: a password containing @, : or / breaks it. URL.create builds it from parts, and the parts come from the environment:

import os
from sqlalchemy import URL, create_engine, text

sqlite_engine = create_engine("sqlite:///shop.db")

my_engine = create_engine(URL.create(
    "mysql+pymysql",
    username=os.environ["MYSQL_USER"],
    password=os.environ["MYSQL_PASSWORD"],
    host=os.environ["MYSQL_HOST"],
    port=int(os.environ["MYSQL_PORT"]),
    database=os.environ["MYSQL_DATABASE"],
))

pg_engine = create_engine(URL.create(
    "postgresql+psycopg2",
    username=os.environ["PGUSER"],
    password=os.environ["PGPASSWORD"],
    host=os.environ["PGHOST"],
    port=int(os.environ["PGPORT"]),
    database=os.environ["PGDATABASE"],
), pool_size=5, max_overflow=10, pool_pre_ping=True)

for engine in (sqlite_engine, my_engine, pg_engine):
    print(f"{engine.dialect.name:10} driver={engine.driver}")

sqlite     driver=pysqlite
mysql      driver=pymysql
postgresql driver=psycopg2

create_engine does not connect yet; the first query does. Create one engine per database when the program starts and share it, never one per query.

Core: SQL you write, run the same way everywhere

SQLAlchemy Core is the layer just above the DB-API. Its text() construct runs your own SQL with :name parameters on every database. SQLAlchemy converts them to the driver's style (? or %s), so this one function works unchanged on all three engines:

def city_customers(engine, city):
    with engine.connect() as conn:                      # borrowed from the pool, returned at the end
        result = conn.execute(
            text("SELECT name, joined_on FROM customers WHERE city = :city ORDER BY customer_id"),
            {"city": city},
        )
        return [(row.name, str(row.joined_on)) for row in result]

for engine in (sqlite_engine, my_engine, pg_engine):
    print(engine.dialect.name, city_customers(engine, "Chattogram"))

sqlite [('Tanvir Ahmed', '2025-12-14')]
mysql [('Tanvir Ahmed', '2025-12-14')]
postgresql [('Tanvir Ahmed', '2025-12-14')]

Rows behave like named tuples (row.name). A connection from engine.connect() never autocommits: a write on it is kept only if you call conn.commit(), and anything uncommitted is rolled back when the block ends. For writes, engine.begin() is simpler: it opens a transaction that commits at the end of the block and rolls back if it raises; unlike the DB-API with con:, the connection also goes back to the pool:

with pg_engine.begin() as conn:
    conn.execute(text("UPDATE products SET stock = stock + :n WHERE product_id = :id"), {"n": 10, "id": 9})

try:
    with pg_engine.begin() as conn:
        conn.execute(text("UPDATE products SET stock = stock + :n WHERE product_id = :id"), {"n": 5, "id": 9})
        conn.execute(text("UPDATE products SET price = 0 WHERE product_id = 9"))       # breaks CHECK (price > 0)
except Exception as exc:
    print(type(exc).__name__, "- rolled back")

print(pd.read_sql(text("SELECT product_id, stock FROM products WHERE product_id = :id"), pg_engine, params={"id": 9}))

IntegrityError - rolled back
   product_id  stock
0           9     10

Stock is 10, not 15: the second block was undone as a whole. SQLAlchemy wraps driver errors in its own classes (sqlalchemy.exc.IntegrityError here), whose .orig attribute holds the driver's original exception.

The same query from three databases

Because read_sql accepts any engine, moving an analysis from your laptop's SQLite to the company's PostgreSQL is a one-word change. Do check the types, though:

revenue_sql = text("""
    SELECT o.status, SUM(oi.quantity * oi.unit_price) AS revenue
    FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
    GROUP BY o.status ORDER BY o.status
""")
for engine in (sqlite_engine, my_engine, pg_engine):
    df = pd.read_sql(revenue_sql, engine)
    print(f"{engine.dialect.name:10} total={df['revenue'].sum()}  dtype={df['revenue'].dtype}")

sqlite     total=103550  dtype=int64
mysql      total=103550.0  dtype=float64
postgresql total=103550  dtype=int64

Same total, different types. MySQL's SUM of integers returns DECIMAL; the driver hands over Python Decimal objects and read_sql converts them to float64 (its coerce_float=True default). For a chart that is harmless. For money, IDs or anything you compare exactly, it is not: a float cannot hold every decimal amount (0.1 is already approximate) or every integer above 2^53 exactly, so exact comparisons and sums of money can drift. Cast in SQL (CAST(SUM(…) AS SIGNED) in MySQL, ::bigint in PostgreSQL) or in pandas (df["revenue"].astype("int64")).

select(): queries built in Python

Core can also build SQL from Python objects. You describe a table once (here SQLAlchemy reads the columns from the database: reflection), then compose queries; SQLAlchemy writes the right SQL for each dialect:

from sqlalchemy import MetaData, Table, func, select
from sqlalchemy.dialects import mssql, postgresql

products = Table("products", MetaData(), autoload_with=sqlite_engine)

stmt = (
    select(products.c.name, products.c.price)
    .where(products.c.price >= 3000)
    .order_by(products.c.price.desc())
    .limit(3)
)
with sqlite_engine.connect() as conn:
    print(conn.execute(stmt).all())

print(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
print(stmt.compile(dialect=mssql.dialect(), compile_kwargs={"literal_binds": True}))

[('27-inch Monitor', 18500), ('AI Engineering Bootcamp', 15000), ('Noise-Cancelling Headphones', 7500)]
SELECT products.name, products.price 
FROM products 
WHERE products.price >= 3000 ORDER BY products.price DESC 
 LIMIT 3
SELECT TOP 3 products.name, products.price 
FROM products 
WHERE products.price >= 3000 ORDER BY products.price DESC

The same Python object became LIMIT 3 for PostgreSQL and TOP 3 for SQL Server. select() shines when a query is assembled from pieces at run time (optional filters from a search form, a column chosen by the user from an allow-list) because you never concatenate strings.

The ORM: rows as Python objects

The ORM (object-relational mapper) maps a class to a table and each object to a row. You declare the model once:

from datetime import date
from sqlalchemy import ForeignKey, String
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)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(100), unique=True)
    city: Mapped[str | None] = mapped_column(String(50))       # "| None" = nullable
    joined_on: Mapped[date]
    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"))
    order_date: Mapped[date]
    status: Mapped[str] = mapped_column(String(20))
    customer: Mapped[Customer] = relationship(back_populates="orders")

with Session(sqlite_engine) as session:
    nadia = session.get(Customer, 1)                                  # by primary key
    print(nadia.name, nadia.joined_on, [(o.order_id, o.status) for o in nadia.orders])

    no_city = session.scalars(select(Customer).where(Customer.city.is_(None))).all()
    print([c.name for c in no_city])

    session.add(Customer(name="Jamal Uddin", email="jamal@example.com",
                         city="Cumilla", joined_on=date(2026, 6, 1)))
    session.commit()                                                  # INSERT happens here
    print(session.scalar(select(func.count()).select_from(Customer)), "customers")

Nadia Rahman 2025-11-03 [(1, 'delivered'), (3, 'delivered'), (10, 'delivered')]
['Sadia Chowdhury']
9 customers

• The tables already exist, so the model only describes them. For a new project, Base.metadata.create_all(engine) creates them, and a migration tool (Alembic) changes them later.
• session.scalars(stmt) is short for session.execute(stmt).scalars(): execute returns rows (each holding one Customer), scalars() unwraps them into the objects themselves.
• Mapped[date] turns SQLite's text dates into real date objects, so the same model works on every database.
• nadia.orders ran a second query when you touched it (lazy loading). Looping over 1,000 customers and touching .orders on each runs 1,001 queries: the N+1 problem from EXPLAIN and Query Optimization. Load related rows in one go with select(Customer).options(selectinload(Customer.orders)).

Connection pools

Each engine owns a pool. engine.connect() borrows a connection and the end of the with block returns it, still open, for the next caller. The settings you passed to pg_engine mean:

print(type(pg_engine.pool).__name__, "| size:", pg_engine.pool.size())
print(pg_engine.pool.status())
pg_engine.dispose()           # close every pooled connection (at shutdown, or after forking)
my_engine.dispose()

QueuePool | size: 5
Pool size: 5  Connections in pool: 1 Current Overflow: -4 Current Checked out connections: 0

One connection was opened by the queries above and now waits in the pool; the overflow of −4 means four more can be opened before the pool reaches its size of 5.

• pool_size=5: keep up to 5 connections open; max_overflow=10: allow 10 extra under load, closed again afterwards.
• pool_pre_ping=True: test a connection before lending it, so a connection the server or a firewall silently dropped is replaced instead of failing your query.
• 5 and 10 are also the defaults. A file-based SQLite engine gets a QueuePool too, but an in-memory one (sqlite://) gets a SingletonThreadPool that keeps one connection per thread, because every new connection to :memory: would be a new, empty database.
• pool_recycle=1800 (worth adding for MySQL): replace connections older than 30 minutes (MySQL closes idle ones after wait_timeout).

Which tool when

Real projects mix them: an ORM-based web app, Core text() for a few hand-tuned reports, and pandas in the notebooks and training pipelines that read the same database.

Common mistakes

1. Forgetting index=False. to_sql writes the DataFrame's index as an extra column called index by default:
   PythonCopy

   scores.to_sql("scratch_scores", con)
   print([row[1] for row in con.execute("PRAGMA table_info(scratch_scores)")])

   OutputCopy

   ['index', 'customer_id', 'churn_score', 'model']

2. Using if_exists="replace" on a real table. It drops the table, and its primary key, constraints, indexes and permissions go with it. Create the table with DDL and append; to update existing rows, upsert them with SQL (ON CONFLICT / ON DUPLICATE KEY UPDATE), from the DataFrame's rows or from a staging table (see ETL and ELT: Building a Reliable Data Pipeline).
3. Loading a whole table to filter it in pandas. pd.read_sql("SELECT * FROM events", con) followed by df[df.customer_id == 3] moves every row over the network. Put the filter in SQL with a parameter.
4. Formatting values into the query string. read_sql(f"... WHERE city = '{city}'") is the same injection hole shown in Security: Roles, SQL Injection, Sensitive Data and Backups. Use params (and text() with :name for engines).
5. Trusting the types without looking. A MySQL sum arrives as float64, an integer column with one NULL becomes float64, SQLite dates arrive as text. Print dtypes after every read and cast before building features.

Try it yourself

1. Easy: read the Electronics products (category_id = 2) into a DataFrame with read_sql and a parameter, sorted by price, with product_id as the index.
2. Medium: with pg_engine and text(), read revenue per month (to_char(o.order_date, 'YYYY-MM')) of non-cancelled orders, add a running_total column with cumsum() in pandas, and write the result to a new PostgreSQL table monthly_revenue with to_sql. Read back the row count.
3. Hard: build with select() (not text) a query giving every customer's name and number of orders, including customers with none (outerjoin, group_by), and check that it returns the same rows on sqlite_engine and pg_engine. Use Table(..., autoload_with=...) for customers and orders.

Answers

# 1. Easy
electronics = pd.read_sql(
    "SELECT product_id, name, price FROM products WHERE category_id = ? ORDER BY price",
    con, params=(2,), index_col="product_id",
)
print(electronics)

# 2. Medium
monthly = pd.read_sql(text("""
    SELECT to_char(o.order_date, 'YYYY-MM') AS month,
           SUM(oi.quantity * oi.unit_price)::bigint AS revenue
    FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
    WHERE o.status <> :cancelled
    GROUP BY month ORDER BY month
"""), pg_engine, params={"cancelled": "cancelled"})
monthly["running_total"] = monthly["revenue"].cumsum()
monthly.to_sql("monthly_revenue", pg_engine, index=False, if_exists="replace")
print(monthly)
print(pd.read_sql(text("SELECT COUNT(*) AS n FROM monthly_revenue"), pg_engine))

# 3. Hard
def orders_per_customer(engine):
    meta = MetaData()
    customers = Table("customers", meta, autoload_with=engine)
    orders_t = Table("orders", meta, autoload_with=engine)
    stmt = (
        select(customers.c.name, func.count(orders_t.c.order_id).label("n_orders"))
        .select_from(customers.outerjoin(orders_t, orders_t.c.customer_id == customers.c.customer_id))
        .where(customers.c.customer_id <= 8)
        .group_by(customers.c.customer_id, customers.c.name)
        .order_by(customers.c.customer_id)
    )
    with engine.connect() as conn:
        return [tuple(row) for row in conn.execute(stmt)]

lite, pg = orders_per_customer(sqlite_engine), orders_per_customer(pg_engine)
print(lite)
print("same on both:", lite == pg)
pg_engine.dispose()

Summary

• pd.read_sql(sql, con_or_engine, params=…, parse_dates=…, index_col=…, chunksize=…) turns a query into a DataFrame (or an iterator of them). Let SQL filter and aggregate; let pandas do the rest.
• df.to_sql(name, con, if_exists=…, index=False, chunksize=…, method="multi") writes a DataFrame. Design important tables yourself and append; replace drops the table.
• A SQLAlchemy engine is built from a URL (use URL.create with values from the environment), owns a connection pool, and is created once per program.
• Core text() runs your SQL portably with :name parameters; select() builds SQL per dialect; the ORM maps classes to tables and rows to objects.
• Always check dtypes: dates from SQLite are text, sums from MySQL become floats.

Next: Data Cleaning with pandas takes a deliberately messy orders file and turns it, step by step, into data you would trust to train a model, using the DataFrames you now know how to get in and out of a database.