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} |