From CSV Files to Live Databases: Connecting Python to Production Data

You have spent the last two lessons loading data with pd.read_csv("sales.csv"). That works in a classroom. On your first real job you will hear this instead: "The orders are in the Postgres warehouse — pull the last 90 days into Python and send me the RFM." No CSV was attached. No file to download. The table lives on a server and you must query it into a DataFrame, analyze it, and write your results back. This article makes that jump concrete — with the same e-commerce data you will use in the notebook and Project 35.

1. Why Your Notebook Needs a Database Connection

Think of a database as a shared Excel file that 200 people and 15 services write to at the same time — but it enforces types, prevents corruption, and answers questions in milliseconds. Your Python notebook is the client that asks questions (SQL) and gets answers (rows) back as a DataFrame.

You want to…	Old CSV way	Production way (this article)
Load orders	`pd.read_csv("orders.csv")`	`pd.read_sql_query("SELECT * FROM orders", engine)`
Filter completed only	Load everything, then `df[df.status=="completed"]`	Filter inside the DB: `WHERE status='completed'` — less RAM, faster
Save your RFM result	`df.to_csv("rfm.csv")` on your laptop	`df.to_sql("rfm_scores", engine, schema="marketing_analytics")` — visible to the whole team

2. The Two Layers Between Python and the Database

Every Python DB connection has two parts. You already used Layer 1 with SQLite — now we add Layer 2 for everything else:

Layer	What it does	Examples	Analogy
1 — DB-API Driver	Speaks the database's native network protocol (TCP, auth, wire format)	`sqlite3` (built-in), `psycopg2` (Postgres), `pymysql` (MySQL)	The phone that dials the warehouse
2 — SQLAlchemy Engine	Manages connection pooling, translates dialects, plugs directly into `pd.read_sql` / `to_sql`	`create_engine("postgresql+psycopg2://...")`	The switchboard that pools calls and translates languages

For SQLite the driver is built into Python (import sqlite3). For Postgres/MySQL you install the driver once with pip install psycopg2-binary or pymysql and SQLAlchemy handles the rest.

3. Your Starting Point: The Same Orders Table — Now Inside Postgres

In the notebook you saw this table in SQLite :memory:. In production it lives at prod-db.company.com in the table orders. Same columns, same 280 rows — just a different address:

transaction_id	customer_id	transaction_date	status	quantity	unit_price
1001	C001	2024-06-22	completed	1	899.00
1002	C002	2024-02-12	completed	5	80.00
1003	C003	2024-06-25	completed	1	45.00
1004	C001	2024-06-28	completed	3	45.00

The code below connects to both — one line changes (the URI), everything else stays identical.

from sqlalchemy import create_engine
import pandas as pd

# Postgres warehouse (production — needs psycopg2 driver)
engine_pg = create_engine("postgresql+psycopg2://read_only_user:SecretPass123@prod-db.company.com:5432/analytics_db")

# SQLite local file (what the videos used — same API, different URI)
engine_sqlite = create_engine("sqlite:///ecommerce_warehouse.db")

# SQLite in-memory (what the notebook uses: no file at all)
import sqlite3
conn = sqlite3.connect(":memory:")

URI pattern to memorize: dialect+driver://user:password@host:port/database — only the prefix changes (postgresql+psycopg2:// vs mysql+pymysql:// vs sqlite:///).

4. Reading: Three Pandas Functions — When to Use Which

All three return a DataFrame. Pick based on how much SQL you want to write:

Function	What you write	Best for	Example
`pd.read_sql_table("orders", engine)`	Just a table name	Dump a whole table, no filter	`df = pd.read_sql_table("orders", engine_pg)`
`pd.read_sql_query("SELECT ...", engine)`	Full SQL with `WHERE/JOIN/GROUP BY`	Filtered, aggregated extracts — use this 90% of the time	`pd.read_sql_query("SELECT customer_id, SUM(quantity*unit_price) AS total FROM orders WHERE status='completed' GROUP BY customer_id", engine_pg)`
`pd.read_sql("SELECT ...", engine)`	Either of the above (auto-detects)	Quick explorations; prefers explicit choice in production	`pd.read_sql(query, engine_pg)`

Realistic filtered extract you will actually run (same idea as Project 35):

query = """
SELECT customer_id, DATE(transaction_date) AS order_date,
       SUM(quantity * unit_price) AS total_spend
FROM orders
WHERE status = 'completed' AND transaction_date >= '2024-01-01'
GROUP BY customer_id, DATE(transaction_date)
"""
df_orders = pd.read_sql_query(query, engine_pg)
print(df_orders.head(3))
#   customer_id  order_date  total_spend
# 0        C001  2024-06-22       899.00
# 1        C001  2024-06-28       135.00
# 2        C002  2024-02-12       400.00

5. Writing Back: Share Your Results With the Team

After engineering RFM you need to publish rfm_scores so Tableau and the marketing team can use it — not leave it on your laptop. That is to_sql:

rfm = df_orders.groupby('customer_id')['total_spend'].sum().reset_index()

rfm.to_sql(
    name='customer_lifetime_spend',     # new table name
    con=engine_pg,
    schema='marketing_analytics',       # Postgres schema (like a folder for tables)
    if_exists='replace',                # 'fail' = error if exists | 'replace' = drop+recreate | 'append' = keep old + add rows
    index=False,                        # don't write the DataFrame's 0,1,2 index as a column
    chunksize=1000                      # write 1000 rows at a time (see next section)
)

`if_exists`	What happens if the table already exists?	When to use
`'fail'`	Raises an error — refuses to overwrite (default, safe)	First-time creation; guard against accidents
`'replace'`	Drops the old table and creates a fresh one	Refreshing a report table daily
`'append'`	Keeps existing rows, inserts new ones	Appending a new day's increment

Why index=False matters: Without it, Pandas adds an index column (0,1,2…) to your table. Every downstream query then has a mystery extra column — easy to forget, annoying to clean.

6. When the Table Is Too Big for RAM: `chunksize`

Your warehouse orders table has 40 million rows — SELECT * into one DataFrame will crash the notebook. Both reading and writing have the same fix: stream in batches.

# Reading: process 5000 rows at a time, never holding all 40M in memory
total = 0
for chunk in pd.read_sql_query("SELECT * FROM orders", engine_pg, chunksize=5000):
    total += chunk['quantity'].sum()   # or filter/aggregate/write each chunk
print(f"Total units sold: {total}")

# Writing: flush your 2M-row DataFrame 1000 rows at a time
rfm.to_sql('customer_lifetime_spend', engine_pg, if_exists='replace', index=False, chunksize=1000)

Quiz Q1-08 tests exactly this. Whenever the prompt says "millions of rows / won't fit in RAM" think chunksize immediately.

7. Production Hygiene: Don't Hard-Code Secrets, Don't Leak Connections

Two mistakes every beginner makes once (and then never again): committing a password to Git, and leaving a connection open until the DB refuses new queries.

# .env file — lives on your machine, never committed (add .env to .gitignore):
# DB_URI=postgresql+psycopg2://read_only_user:SecretPass123@prod-db.company.com:5432/analytics_db

from dotenv import load_dotenv
import os
load_dotenv()  # reads .env into environment variables

engine = create_engine(
    os.getenv("DB_URI"),
    pool_size=5, max_overflow=10,    # keep 5 connections warm, burst to 15 under load
    pool_pre_ping=True               # test that a pooled connection is still alive before using it
)

# Use the engine, then close it when the notebook is done:
# engine.dispose()

# For user-supplied filters, never use f-strings — use bound parameters:
from sqlalchemy import text
with engine.connect() as conn:   # auto-closes even if the query fails
    result = conn.execute(text("SELECT * FROM orders WHERE status = :s"), {"s": "completed"})
    df = pd.DataFrame(result.fetchall(), columns=result.keys())
print(df.head(2))

Mistake	What happens	Fix you just saw
Hard-code `"postgresql://user:pass@..."` in the notebook	Password leaks to GitHub; key rotation breaks the notebook	`os.getenv("DB_URI")` + `.env` + `.gitignore`
Leave `engine` open forever	Pool exhausts; next `read_sql` hangs	`with engine.connect() as conn:` or `engine.dispose()`
`f"SELECT ... WHERE status='{user_input}'"`	SQL injection — user can type `'; DROP TABLE orders; --`	`text("... WHERE status=:s")` with `{"s": user_input}`