Reading from SQL Databases into pandas with SQLAlchemy

Introduction

Picture this: your company's entire sales history lives in a PostgreSQL database. You need to pull the last 18 months of transactions, join them against a product table, filter to three specific regions, and hand the result to a pandas pipeline that calculates regional revenue trends. Your current process involves exporting a CSV from the reporting tool, cleaning it up in Excel, and then loading it into Python. That's four manual steps, each introducing a chance for human error, and every time the data changes you do it again.

There is a better way. The combination of SQLAlchemy and pandas gives you a direct bridge from any relational database — PostgreSQL, MySQL, SQL Server, SQLite, Oracle, BigQuery — into a DataFrame, with no CSV intermediary, no manual exports, and no data corruption from ill-formatted spreadsheet exports. You write a query, you get a DataFrame. You can parameterize it, schedule it, test it, and version-control it like any other piece of code.

By the end of this lesson you'll know how to build that bridge properly. We're not just going to call pd.read_sql and move on. We're going to understand the architecture of database connections, how SQLAlchemy's engine and connection model works, when to use raw SQL versus ORM queries, how to handle credentials securely, and how to optimize for performance when you're pulling millions of rows. This is production-grade database reading, not tutorial-level demos.

What you'll learn:

• How SQLAlchemy's engine, connection pool, and dialect system work, and why that architecture matters for data analysis
• How to construct engines for PostgreSQL, MySQL, SQLite, and SQL Server with proper connection string patterns
• The three distinct approaches to reading SQL into pandas (read_sql, read_sql_query, read_sql_table) and when each is appropriate
• How to parameterize queries safely without SQL injection risks
• Performance optimization strategies: chunking, server-side cursors, type hints, and reducing data transfer
• Secure credential management patterns for production environments
• Troubleshooting the most common connection and data-type errors you will actually encounter

---

Prerequisites

This lesson assumes you're comfortable with Python fundamentals — if you want a refresher on Python data structures and control flow, see Python Basics for Excel Users: Variables, Lists, Dictionaries, and Loops. You should also be familiar with basic pandas operations like loading data and exploring DataFrames — Your First pandas DataFrame: Loading CSV and Excel Files and Exploring Data covers that ground. Basic SQL — SELECT, JOIN, WHERE, GROUP BY — is assumed. Finally, make sure your environment is set up; Setting Up Python for Data Analysis: Python, VS Code, Jupyter, and Virtual Environments walks through the tooling if needed.

Install the required packages:

pip install sqlalchemy pandas pyarrow
# For PostgreSQL:
pip install psycopg2-binary
# For MySQL:
pip install pymysql
# For SQL Server:
pip install pyodbc

---

How SQLAlchemy Actually Works (and Why You Should Care)

Most tutorials treat SQLAlchemy as a magic connection string. That's a mistake. Understanding the architecture will save you hours of debugging mysterious connection errors and performance problems.

SQLAlchemy has two major layers:

The Core layer deals with SQL expression constructs, schema definitions, and connection management. This is what you'll use for data analysis work — it handles engines, connections, and executing raw or semi-structured SQL.

The ORM layer maps Python classes to database tables and is primarily a web application tool. You'll occasionally encounter ORM constructs when integrating with an existing application's models, but for analytical work you almost never need it.

The key objects in your workflow are:

Engine → Connection Pool → Connection → Result Set → DataFrame

The Engine is a factory for connections. You create it once at module or application startup and reuse it for the lifetime of your session. It does not itself hold an open database connection — it creates and manages a pool of them.

The Connection Pool is where the real magic happens. By default, SQLAlchemy maintains a pool of open database connections so that each query doesn't pay the overhead of a fresh TCP handshake, authentication, and session setup. For PostgreSQL, that handshake can take 50–100ms — a real cost if you're running dozens of queries in a pipeline.

The Dialect is the database-specific translation layer. When you write engine.connect(), SQLAlchemy uses the dialect to speak the correct wire protocol for your database. The dialect knows about PostgreSQL-specific RETURNING clauses, MySQL's LIMIT syntax differences, and SQL Server's TOP N versus LIMIT. You benefit from this transparently.

Here's how to think about the connection string URL that defines an engine:

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

Breaking it down:

• dialect: postgresql, mysql, mssql, sqlite, oracle
• driver: the Python DBAPI library (psycopg2, pymysql, pyodbc)
• The rest: standard connection parameters

---

Building Engines for Real Databases

SQLite: Starting Simple

SQLite needs no server and is perfect for local data files, testing, and embedded analytics:

from sqlalchemy import create_engine

# File-based SQLite database
engine = create_engine("sqlite:///data/sales.db")

# In-memory SQLite (useful for tests)
engine_mem = create_engine("sqlite:///:memory:")

Three slashes means relative path; four slashes means absolute path on Unix:

engine_abs = create_engine("sqlite:////home/user/data/sales.db")

PostgreSQL

engine = create_engine(
    "postgresql+psycopg2://analyst:s3cr3tpassword@db.company.com:5432/analytics_prod",
    pool_size=5,          # Keep 5 connections open
    max_overflow=10,      # Allow up to 10 additional connections under load
    pool_timeout=30,      # Wait up to 30 seconds for a connection from the pool
    pool_pre_ping=True,   # Test connection health before use
)

pool_pre_ping=True deserves special mention. Without it, if your database server closes an idle connection (common with cloud databases that have idle connection timeouts), your next query will fail with a cryptic "connection was closed" error rather than automatically recovering. With pool_pre_ping, SQLAlchemy sends a lightweight probe before handing you a connection, and if it's dead, it recycles and gives you a fresh one.

MySQL

engine = create_engine(
    "mysql+pymysql://analyst:s3cr3tpassword@db.company.com:3306/analytics",
    connect_args={"charset": "utf8mb4"},  # Handle full Unicode including emoji
)

SQL Server

import urllib

params = urllib.parse.quote_plus(
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=db.company.com;"
    "DATABASE=analytics_prod;"
    "UID=analyst;"
    "PWD=s3cr3tpassword;"
)
engine = create_engine(f"mssql+pyodbc:///?odbc_connect={params}")

The URL-encoding step is required because SQL Server connection strings use semicolons and spaces that would break the SQLAlchemy URL parser.

For Windows integrated authentication (common in corporate environments):

params = urllib.parse.quote_plus(
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=db.company.com;"
    "DATABASE=analytics_prod;"
    "Trusted_Connection=yes;"
)
engine = create_engine(f"mssql+pyodbc:///?odbc_connect={params}")

---

Secure Credential Management

Hardcoding passwords in your scripts is a career-limiting move in professional environments. Here are the patterns that actually hold up in production.

Environment Variables

The minimum viable approach:

import os
from sqlalchemy import create_engine

DB_URL = (
    f"postgresql+psycopg2://"
    f"{os.environ['DB_USER']}:{os.environ['DB_PASSWORD']}"
    f"@{os.environ['DB_HOST']}:{os.environ.get('DB_PORT', '5432')}"
    f"/{os.environ['DB_NAME']}"
)
engine = create_engine(DB_URL)

Set them in your shell before running:

export DB_USER=analyst
export DB_PASSWORD=s3cr3tpassword
export DB_HOST=db.company.com
export DB_NAME=analytics_prod

Or in a .env file with python-dotenv:

from dotenv import load_dotenv
load_dotenv()  # Reads .env file into os.environ

engine = create_engine(os.environ["DATABASE_URL"])

Make sure .env is in your .gitignore. Always.

SQLAlchemy URL with `create_engine` and Secret Managers

For production pipelines deployed in AWS, Azure, or GCP, pull credentials from the platform's secret manager at runtime:

import boto3
import json
from sqlalchemy import create_engine

def get_db_engine():
    client = boto3.client("secretsmanager", region_name="us-east-1")
    secret = json.loads(
        client.get_secret_value(SecretId="prod/analytics/db")["SecretString"]
    )
    url = (
        f"postgresql+psycopg2://{secret['username']}:{secret['password']}"
        f"@{secret['host']}:{secret['port']}/{secret['dbname']}"
    )
    return create_engine(url, pool_pre_ping=True)

---

The Three Ways to Read SQL into pandas

pandas offers three functions, and the differences matter more than you'd think.

`pd.read_sql_table`: For Full Table Reads

import pandas as pd

df = pd.read_sql_table(
    "dim_product",
    con=engine,
    schema="public",      # PostgreSQL schema, or dbo for SQL Server
    columns=["product_id", "product_name", "category", "unit_cost"],
)

read_sql_table reads an entire table without you writing any SQL. It's appropriate for small dimension tables — products, customers, regions — where you want everything. It does not accept a WHERE clause. For filtered reads, you need read_sql_query.

`pd.read_sql_query`: For SQL Queries

This is what you'll use 90% of the time:

query = """
    SELECT
        t.transaction_id,
        t.transaction_date,
        t.amount,
        t.region_code,
        p.product_name,
        p.category
    FROM fact_transactions t
    JOIN dim_product p ON t.product_id = p.product_id
    WHERE t.transaction_date >= '2023-01-01'
      AND t.region_code IN ('AMER', 'EMEA', 'APAC')
    ORDER BY t.transaction_date
"""

df = pd.read_sql_query(query, con=engine)

The result is a DataFrame with column names taken from the SQL result set aliases. The ORDER BY in SQL is preserved in the DataFrame row order, which matters for time-series work.

`pd.read_sql`: The Polymorphic Version

read_sql accepts either a table name or a SQL string and dispatches to the appropriate underlying function. It's convenient for interactive use but makes code less explicit:

# These are equivalent:
df = pd.read_sql("SELECT * FROM dim_product", con=engine)
df = pd.read_sql("dim_product", con=engine)  # Full table read

In production code, prefer read_sql_query or read_sql_table because they're explicit about intent.

---

Using Connections and Transactions for Read Consistency

When you're reading multiple related tables for a join or analysis, you want all reads to reflect the same database snapshot — no rows appearing in one table that haven't appeared in the other yet. Use a connection context manager:

with engine.connect() as conn:
    df_transactions = pd.read_sql_query(
        "SELECT * FROM fact_transactions WHERE transaction_date >= '2024-01-01'",
        con=conn
    )
    df_adjustments = pd.read_sql_query(
        "SELECT * FROM fact_adjustments WHERE adjustment_date >= '2024-01-01'",
        con=conn
    )

# Both DataFrames reflect the same consistent database snapshot

The connection is automatically returned to the pool when the with block exits, whether normally or via exception. This is critical for connection pool health — leaked connections eventually exhaust the pool and cause your pipeline to hang.

For databases with transaction isolation guarantees (PostgreSQL, SQL Server), wrapping reads in an explicit transaction further strengthens consistency:

with engine.begin() as conn:
    # All reads inside this block see a consistent snapshot (READ COMMITTED or REPEATABLE READ)
    df_orders = pd.read_sql_query("SELECT * FROM orders WHERE status = 'pending'", con=conn)
    df_inventory = pd.read_sql_query("SELECT * FROM inventory WHERE quantity > 0", con=conn)

---

Safe Query Parameterization

Never build queries by string interpolation. Never. Not even for "internal" tools that "only you use."

The wrong way:

region = "AMER"  # Imagine this comes from user input
query = f"SELECT * FROM transactions WHERE region = '{region}'"
# SQL injection risk — don't do this

The right way — using bound parameters:

SQLAlchemy Core uses :param_name placeholders and a separate parameters dictionary:

from sqlalchemy import text

query = text("""
    SELECT
        t.transaction_id,
        t.transaction_date,
        t.amount,
        p.product_name
    FROM fact_transactions t
    JOIN dim_product p ON t.product_id = p.product_id
    WHERE t.region_code = :region
      AND t.transaction_date BETWEEN :start_date AND :end_date
      AND t.amount > :min_amount
""")

with engine.connect() as conn:
    df = pd.read_sql_query(
        query,
        con=conn,
        params={
            "region": "AMER",
            "start_date": "2024-01-01",
            "end_date": "2024-06-30",
            "min_amount": 500.0,
        }
    )

The text() wrapper tells SQLAlchemy to treat the string as a SQL expression with named placeholders, and the database driver handles proper escaping. The region "AMER'; DROP TABLE transactions; --" becomes a literal string value, not executable SQL.

Handling IN clauses with multiple values:

IN clauses are trickier because you can't bind a list to a single placeholder directly. The cleanest approach:

from sqlalchemy import text, bindparam

regions = ["AMER", "EMEA", "APAC"]

query = text(
    "SELECT * FROM transactions WHERE region_code IN :regions"
).bindparams(bindparam("regions", expanding=True))

with engine.connect() as conn:
    df = pd.read_sql_query(query, con=conn, params={"regions": regions})

The expanding=True argument tells SQLAlchemy to expand the list into properly counted placeholders at query execution time — (:regions_1, :regions_2, :regions_3) — while still binding values safely.

---

Type Handling: From SQL Types to pandas dtypes

One of the underappreciated challenges of SQL-to-pandas pipelines is data type translation. Not everything maps cleanly.

Dates and Timestamps

Most databases return timestamps as Python datetime objects, which pandas correctly infers as datetime64[ns]. However, timezone-aware timestamps require explicit handling:

# PostgreSQL TIMESTAMPTZ returns timezone-aware datetimes
df = pd.read_sql_query(
    "SELECT transaction_id, created_at FROM transactions LIMIT 5",
    con=engine
)
print(df.dtypes)
# created_at    datetime64[ns, UTC]  ← timezone-aware if DB stores UTC

If you need to convert to a local timezone for display:

df["created_at_local"] = df["created_at"].dt.tz_convert("America/New_York")

For more on time-series work in pandas, see Working with Dates and Time Series in pandas: Parsing, Resampling, and Rolling Windows.

Nullable Integers

SQL's INTEGER columns can contain NULL. In older pandas (pre-1.0), this forced the entire column to float64 because Python's int can't represent None. Modern pandas has nullable integer types:

df = pd.read_sql_query(query, con=engine)

# Convert nullable integer columns explicitly
df["customer_id"] = df["customer_id"].astype("Int64")  # Capital I — nullable

Decimal vs Float

Database DECIMAL/NUMERIC types often come through as Python Decimal objects (from the decimal module), which pandas stores as object dtype. This kills performance and breaks arithmetic. Convert explicitly:

df["unit_price"] = pd.to_numeric(df["unit_price"], errors="coerce").astype("float64")

Alternatively, use dtype argument in read_sql_query to specify column types upfront:

df = pd.read_sql_query(
    query,
    con=engine,
    dtype={"unit_price": "float64", "quantity": "Int64"}
)

---

Performance Optimization for Large Result Sets

The Chunking Pattern

When you need to read millions of rows, loading them all into memory at once isn't always feasible. read_sql_query supports a chunksize parameter that returns an iterator of DataFrames:

chunk_iter = pd.read_sql_query(
    "SELECT * FROM fact_transactions WHERE fiscal_year = 2023",
    con=engine,
    chunksize=50_000
)

processed_chunks = []
for chunk in chunk_iter:
    # Process each chunk — aggregate, filter, transform
    chunk["revenue"] = chunk["quantity"] * chunk["unit_price"]
    agg = chunk.groupby("product_category")["revenue"].sum()
    processed_chunks.append(agg)

final = pd.concat(processed_chunks).groupby(level=0).sum()

This processes 50,000 rows at a time rather than loading the full dataset. Memory usage stays bounded regardless of total result size.

Server-Side Cursors (PostgreSQL)

By default, PostgreSQL fetches all rows from a query result into client memory before returning anything. For very large results, this is both slow (you wait for everything to arrive) and memory-intensive. Server-side cursors stream rows on demand:

with engine.connect().execution_options(stream_results=True) as conn:
    df = pd.read_sql_query(
        "SELECT * FROM fact_transactions WHERE fiscal_year = 2023",
        con=conn,
        chunksize=10_000,
    )
    for chunk in df:
        process(chunk)

stream_results=True tells the psycopg2 driver to use a named server-side cursor. Rows are fetched in batches as you iterate. This is the right approach for multi-million-row reads where even the network transfer time matters.

Column Selection and Pushdown

The single most effective performance optimization is selecting only the columns you need. A SELECT * on a wide table (say, 80 columns) transfers 80x the data of a targeted SELECT col1, col2, col3. This sounds obvious, but it's startlingly common in real data pipelines.

# Bad: fetches 80 columns, you use 6
df = pd.read_sql_query("SELECT * FROM fact_transactions", con=engine)
df = df[["transaction_id", "transaction_date", "amount", "region", "product_id", "channel"]]

# Good: fetches exactly what you need
df = pd.read_sql_query("""
    SELECT transaction_id, transaction_date, amount, region, product_id, channel
    FROM fact_transactions
""", con=engine)

Similarly, push filtering into SQL rather than loading then filtering in pandas. Every row your WHERE clause eliminates is a row that doesn't cross the network.

Using `parse_dates` for Automatic Date Parsing

Date parsing in pandas adds overhead if done column-by-column after loading. The parse_dates argument handles it during the read:

df = pd.read_sql_query(
    query,
    con=engine,
    parse_dates=["transaction_date", "created_at", "updated_at"]
)

Alternatively, if you have many timestamp columns, you can pass a dict with format hints:

df = pd.read_sql_query(
    query,
    con=engine,
    parse_dates={"transaction_date": "%Y-%m-%d", "created_at": "%Y-%m-%d %H:%M:%S"}
)

Index Columns

If you're going to be doing a lot of label-based lookups after loading, you can specify which column becomes the DataFrame index at read time:

df = pd.read_sql_query(
    "SELECT product_id, product_name, category, unit_cost FROM dim_product",
    con=engine,
    index_col="product_id"
)

# Now lookups are O(1) hash operations rather than O(n) scans
unit_cost = df.loc["PRD-00447", "unit_cost"]

This is particularly useful for dimension tables you'll use repeatedly for lookups or merges.

---

Reflecting Existing Database Schemas

When you're connecting to an existing database and you need to introspect its structure — what tables exist, what columns they have, what types — SQLAlchemy's reflection capability is invaluable.

from sqlalchemy import inspect

inspector = inspect(engine)

# List all tables in the database
tables = inspector.get_table_names(schema="public")
print(tables)
# ['dim_customer', 'dim_product', 'dim_region', 'fact_transactions', 'fact_adjustments']

# Get column details for a specific table
columns = inspector.get_columns("fact_transactions", schema="public")
for col in columns:
    print(f"{col['name']:30s} {str(col['type']):20s} nullable={col['nullable']}")

Output:

transaction_id                 INTEGER              nullable=False
transaction_date               DATE                 nullable=False
amount                         NUMERIC(12, 2)       nullable=True
region_code                    VARCHAR(10)          nullable=True
product_id                     INTEGER              nullable=True
channel                        VARCHAR(50)          nullable=True

This is extremely useful when you're onboarding to an unfamiliar database and need to understand what you're working with before writing queries.

You can also get foreign key relationships:

fks = inspector.get_foreign_keys("fact_transactions")
for fk in fks:
    print(f"Column {fk['constrained_columns']} → {fk['referred_table']}.{fk['referred_columns']}")

---

Building Reusable Query Functions

In a real data pipeline, you don't write ad-hoc queries scattered through your code. You build functions that encapsulate data access logic, accept parameters, and return clean DataFrames. Here's a production-quality pattern:

from sqlalchemy import create_engine, text
import pandas as pd
from datetime import date
from typing import Optional, List

# Engine created once at module level
engine = create_engine(
    "postgresql+psycopg2://analyst:password@db.company.com/analytics",
    pool_pre_ping=True,
    pool_size=3,
)

def get_regional_transactions(
    regions: List[str],
    start_date: date,
    end_date: date,
    min_amount: Optional[float] = None,
    categories: Optional[List[str]] = None,
) -> pd.DataFrame:
    """
    Load regional transaction data for analysis.
    
    Parameters
    ----------
    regions : list of str
        Region codes to include (e.g., ['AMER', 'EMEA'])
    start_date : date
        Inclusive start of date range
    end_date : date
        Inclusive end of date range
    min_amount : float, optional
        If provided, exclude transactions below this amount
    categories : list of str, optional
        If provided, filter to these product categories only
    
    Returns
    -------
    pd.DataFrame
        Transactions with product and region details, sorted by date
    """
    base_query = """
        SELECT
            t.transaction_id,
            t.transaction_date,
            t.amount,
            t.region_code,
            r.region_name,
            p.product_id,
            p.product_name,
            p.category
        FROM fact_transactions t
        JOIN dim_product p ON t.product_id = p.product_id
        JOIN dim_region r ON t.region_code = r.region_code
        WHERE t.region_code IN :regions
          AND t.transaction_date BETWEEN :start_date AND :end_date
    """
    
    params = {
        "regions": tuple(regions),  # IN clause needs a tuple for psycopg2
        "start_date": start_date,
        "end_date": end_date,
    }
    
    if min_amount is not None:
        base_query += " AND t.amount >= :min_amount"
        params["min_amount"] = min_amount
    
    if categories is not None:
        base_query += " AND p.category IN :categories"
        params["categories"] = tuple(categories)
    
    base_query += " ORDER BY t.transaction_date"
    
    with engine.connect() as conn:
        df = pd.read_sql_query(
            text(base_query),
            con=conn,
            params=params,
            parse_dates=["transaction_date"],
            dtype={"amount": "float64"},
        )
    
    return df


# Usage
df = get_regional_transactions(
    regions=["AMER", "EMEA"],
    start_date=date(2024, 1, 1),
    end_date=date(2024, 6, 30),
    min_amount=1000.0,
    categories=["Electronics", "Software"],
)

This pattern encapsulates all the complexity — parameterization, type casting, connection management, column parsing — behind a clean interface. When your database schema changes, you fix it in one place. When you need to add a new filter, you add one parameter and one condition.

For grouping and aggregating this data or selecting and filtering further after loading, you work with a DataFrame that's already clean and typed correctly.

---

Async Database Access (Advanced Pattern)

For applications that make concurrent database calls — API backends, parallel data pipelines — synchronous SQLAlchemy blocks the event loop. SQLAlchemy 1.4+ includes async support via AsyncEngine:

import asyncio
import pandas as pd
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async_engine = create_async_engine(
    "postgresql+asyncpg://analyst:password@db.company.com/analytics",
    pool_size=10,
    pool_pre_ping=True,
)

async def fetch_region_data(region: str) -> pd.DataFrame:
    async with async_engine.connect() as conn:
        result = await conn.execute(
            text("SELECT * FROM fact_transactions WHERE region_code = :region"),
            {"region": region}
        )
        rows = result.fetchall()
        return pd.DataFrame(rows, columns=result.keys())

async def fetch_all_regions() -> dict:
    regions = ["AMER", "EMEA", "APAC", "LATAM"]
    tasks = [fetch_region_data(r) for r in regions]
    results = await asyncio.gather(*tasks)
    return dict(zip(regions, results))

# All four regions fetched concurrently
dfs = asyncio.run(fetch_all_regions())

Note that pd.read_sql_query doesn't support async connections directly — you need to execute and fetch manually, then construct the DataFrame from the result. For pure analytical scripts this complexity rarely pays off; it's most valuable in production services making parallel database calls.

---

Hands-On Exercise

You'll build a complete data retrieval and analysis pipeline using a real-world scenario. We'll use SQLite so you don't need a server, but the patterns apply identically to PostgreSQL or SQL Server.

Setup: Create the Practice Database

import pandas as pd
from sqlalchemy import create_engine, text
import numpy as np

engine = create_engine("sqlite:///exercise_sales.db")

# Seed data
np.random.seed(42)
n = 10_000

products = pd.DataFrame({
    "product_id": range(1, 51),
    "product_name": [f"Product-{i:03d}" for i in range(1, 51)],
    "category": np.random.choice(["Electronics", "Software", "Hardware", "Services"], 50),
    "unit_cost": np.round(np.random.uniform(10, 500, 50), 2),
})

transactions = pd.DataFrame({
    "transaction_id": range(1, n + 1),
    "transaction_date": pd.date_range("2023-01-01", periods=n, freq="1h").strftime("%Y-%m-%d"),
    "product_id": np.random.randint(1, 51, n),
    "quantity": np.random.randint(1, 20, n),
    "region": np.random.choice(["AMER", "EMEA", "APAC", "LATAM"], n),
    "discount_pct": np.round(np.random.uniform(0, 0.3, n), 3),
})

# Add NULL values to simulate real data messiness
transactions.loc[np.random.choice(n, 200), "discount_pct"] = None

with engine.begin() as conn:
    products.to_sql("dim_product", conn, if_exists="replace", index=False)
    transactions.to_sql("fact_transactions", conn, if_exists="replace", index=False)

print("Database created successfully")

Exercise Tasks

Task 1: Write a parameterized function that accepts a list of regions and a date range, and returns all transactions with product details joined in. Test it with regions=["AMER", "EMEA"] and the full year 2023.

Task 2: Use chunked reading to process the full fact_transactions table in chunks of 1,000 rows. In each chunk, calculate revenue = quantity * unit_cost * (1 - discount_pct). Handle the NULL discount values by treating them as 0% discount. Concatenate the results into a final DataFrame.

Task 3: Using inspect(), list all columns in the fact_transactions table and their types. Then load only the columns you actually need for a revenue analysis (no more than 5 columns).

Task 4: Build a get_category_summary() function that queries the data and returns a DataFrame indexed by category and region showing total revenue, transaction count, and average discount. The aggregation should happen in SQL via GROUP BY, not in pandas.

Sample Solution for Task 4:

from sqlalchemy import text

def get_category_summary() -> pd.DataFrame:
    query = text("""
        SELECT
            p.category,
            t.region,
            COUNT(t.transaction_id) AS transaction_count,
            ROUND(SUM(t.quantity * p.unit_cost * (1 - COALESCE(t.discount_pct, 0))), 2) AS total_revenue,
            ROUND(AVG(COALESCE(t.discount_pct, 0)) * 100, 2) AS avg_discount_pct
        FROM fact_transactions t
        JOIN dim_product p ON t.product_id = p.product_id
        GROUP BY p.category, t.region
        ORDER BY p.category, t.region
    """)
    
    with engine.connect() as conn:
        df = pd.read_sql_query(
            query,
            con=conn,
            dtype={
                "transaction_count": "Int64",
                "total_revenue": "float64",
                "avg_discount_pct": "float64",
            }
        )
    
    return df.set_index(["category", "region"])

summary = get_category_summary()
print(summary)

---

Common Mistakes & Troubleshooting

"Could not connect to server" / Connection Refused

Symptom: sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not connect to server

Causes and fixes:

• Wrong host or port — double-check your connection string
• Database server not running — verify with your DBA
• Firewall blocking the port — especially common for cloud databases; your IP must be whitelisted
• SSL required but not configured — add sslmode=require to the connection args:

engine = create_engine(url, connect_args={"sslmode": "require"})

"Could not parse rfc1738 URL from string"

Symptom: ArgumentError: Could not parse rfc1738 URL from string

Cause: Special characters in your password (like @, #, %) are breaking the URL parser.

Fix: URL-encode the password:

from urllib.parse import quote_plus
password = quote_plus("p@ssw#rd!")  # Encodes special chars
url = f"postgresql+psycopg2://user:{password}@host/db"

DataFrame Loads Successfully but All Values Are None / NaN

Symptom: Query returns rows, DataFrame has right shape, but all values are NaN.

Cause: Column name case mismatch. Some databases (particularly PostgreSQL) return lowercase column names; your query might have them uppercased in aliases.

Fix: Check df.columns.tolist() immediately after loading. Use .lower() or .str.lower() to normalize if needed.

Memory Error on Large Reads

Symptom: MemoryError or process gets killed when reading large tables.

Fix: Use chunksize with an iterator pattern, and ensure you're aggregating within chunks rather than concatenating all chunks (which defeats the purpose):

# Wrong: stores all chunks, uses lots of memory
chunks = list(pd.read_sql_query(query, con=engine, chunksize=10_000))
df = pd.concat(chunks)  # You've loaded everything anyway

# Right: process and reduce each chunk
result = []
for chunk in pd.read_sql_query(query, con=engine, chunksize=10_000):
    result.append(chunk.groupby("category")["revenue"].sum())
final = pd.concat(result).groupby(level=0).sum()

Slow Queries Despite Correct Code

Symptom: Your Python code is fast but the query takes minutes.

Cause: Database-side issue — missing index, full table scan, or poor query plan.

Diagnostic: Run the query with EXPLAIN ANALYZE (PostgreSQL) or SET STATISTICS IO ON (SQL Server) directly in your database client. Look for Seq Scan on large tables where you'd expect index access.

Fix: Work with your DBA to add appropriate indexes. As an analyst, you can suggest: indexes on the columns in your WHERE clause and JOIN conditions. An index on fact_transactions(transaction_date) and fact_transactions(region_code) would dramatically speed up the queries in this lesson.

Decimal Columns Load as Object dtype

Symptom: A column that should be numeric has dtype: object.

Cause: Database returned Decimal objects (Python's decimal.Decimal), which pandas stores as generic Python objects rather than numpy floats.

Fix:

df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

Or specify the dtype at read time:

df = pd.read_sql_query(query, con=engine, dtype={"amount": "float64"})

Connection Pool Exhaustion

Symptom: Application hangs or raises TimeoutError after running correctly for a while.

Cause: Connections aren't being returned to the pool. Common culprit is calling engine.connect() without a context manager:

# Wrong: connection never returned to pool
conn = engine.connect()
df = pd.read_sql_query(query, con=conn)
# conn is never closed!

# Right: context manager guarantees return
with engine.connect() as conn:
    df = pd.read_sql_query(query, con=conn)

Restart the process to clear leaked connections, then fix the code. pool_pre_ping=True helps detect and recover from dead connections but won't fix actively leaked ones.

---

Summary & Next Steps

You now have a complete, production-grade understanding of reading from SQL databases into pandas. Let's consolidate the key principles:

Architecture first: The SQLAlchemy engine is a connection pool factory. Create it once, reuse it everywhere. Use connection context managers (with engine.connect() as conn:) to guarantee connection return.

Right tool for the job: read_sql_query for parameterized SQL queries, read_sql_table for complete dimension tables, and explicit text() wrappers with bound parameters to prevent SQL injection.

Push work to the database: Filter in WHERE, aggregate in GROUP BY, join in JOIN. Every row and column you don't fetch is network traffic and memory you don't spend. Python and pandas are for the work databases can't do as elegantly — reshaping, custom transformations, visualization.

Types matter from the start: Check df.dtypes immediately after loading. Fix Decimal → float64, nullable integers, and timezone-aware timestamps before they corrupt downstream calculations.

Security is non-negotiable: Credentials in environment variables or secret managers, never hardcoded. Parameterized queries always, string interpolation never.

From here, the natural next steps in your analysis pipeline are: cleaning any messy data you've identified in your SQL results, grouping and aggregating to build summary tables, and ultimately visualizing the results in charts that communicate your findings. The database connection pattern you've built here is the foundation every subsequent step builds on.