Table of Contents
- The Python Database Landscape
- SQLAlchemy — The Universal Interface
- pyodbc — Direct SQL Server and Azure SQL Access
- psycopg2 — Native PostgreSQL Access
- sqlite3 — Built-in Local Database
- pandas read_sql() and to_sql() — DataFrame to Database Bridge
- Connection Pooling
- Transactions — All or Nothing
- Storing Credentials Securely with Environment Variables
- Connecting to Azure SQL with Managed Identity
- Production Pattern: Reusable Database Connection Manager
- Common Mistakes
- Interview Questions
- Wrapping Up
Our File Formats post covered reading and writing files. But data engineers spend just as much time reading from and writing to databases — SQL Server, PostgreSQL, Azure SQL, MySQL, and SQLite. This post covers every connection pattern you need: building connection strings, creating engines, executing queries, loading results into DataFrames, writing DataFrames back to tables, handling transactions, and connecting securely to cloud databases.
Analogy — A multilingual translator at a United Nations conference. Each country’s delegate (database) speaks a different language (protocol). The translator (SQLAlchemy) understands all of them and converts every conversation into one common language (Python). You could also speak directly to each delegate in their native tongue (pyodbc for SQL Server, psycopg2 for PostgreSQL) — faster for simple conversations, but you need to learn each language separately. SQLAlchemy gives you one interface for all databases. The native drivers give you raw speed and database-specific features.
The Python Database Landscape
Python has multiple libraries for database access, each serving a different purpose. Here is how they relate to each other and when to use each one.
| Library | What It Does | Best For |
|---|---|---|
| **SQLAlchemy** | Universal database toolkit (ORM + Core) | Any database, production pipelines, portability |
| **pyodbc** | ODBC driver interface for SQL Server/Azure SQL | SQL Server, Azure SQL, Synapse, direct low-level access |
| **psycopg2** | Native PostgreSQL adapter | PostgreSQL, fastest Postgres access from Python |
| **sqlite3** | Built-in SQLite (no install needed) | Local development, testing, embedded databases |
| **mysql-connector-python** | MySQL/MariaDB connector | MySQL, MariaDB |
| **pandas (read_sql/to_sql)** | DataFrame-to-database bridge | Loading query results into DataFrames, bulk inserts |
The relationship: SQLAlchemy sits on top of the native drivers. When you use SQLAlchemy with SQL Server, it calls pyodbc underneath. When you use it with PostgreSQL, it calls psycopg2. You can use the native drivers directly (faster, database-specific) or through SQLAlchemy (portable, higher-level).
SQLAlchemy — The Universal Interface
SQLAlchemy is the standard way to connect Python to databases in production. It provides two layers: Core (SQL expression language — you write SQL but in Python syntax) and ORM (Object-Relational Mapping — you interact with Python classes instead of tables). For data engineering, Core is what you use 95% of the time.
Analogy — A universal power adapter. When you travel internationally, you carry one adapter that works in every country. SQLAlchemy is that adapter — one API that works with SQL Server, PostgreSQL, MySQL, SQLite, Oracle, and 20+ other databases. The connection string is the only thing that changes.
Connection String Formats
The connection string tells SQLAlchemy which database to connect to. The format is dialect+driver://username:password@host:port/database.
from sqlalchemy import create_engine
# SQLite (no server, file-based -- great for local dev)
engine = create_engine("sqlite:///local.db") # File in current directory
engine = create_engine("sqlite:///data/app.db") # Relative path
engine = create_engine("sqlite:////tmp/app.db") # Absolute path (note 4 slashes)
engine = create_engine("sqlite:///:memory:") # In-memory (testing)
# PostgreSQL (psycopg2 is the default driver)
engine = create_engine("postgresql://user:password@localhost:5432/mydb")
engine = create_engine("postgresql+psycopg2://user:password@localhost:5432/mydb") # Explicit driver
# MySQL
engine = create_engine("mysql+pymysql://user:password@localhost:3306/mydb")
# SQL Server (pyodbc)
engine = create_engine(
"mssql+pyodbc://user:password@server:1433/mydb?driver=ODBC+Driver+18+for+SQL+Server"
)
# Azure SQL (with pyodbc connection string)
from urllib.parse import quote_plus
conn_str = (
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=myserver.database.windows.net;"
"DATABASE=mydb;"
"UID=sqladmin;"
"PWD=MyPassword123;"
"Encrypt=yes;"
"TrustServerCertificate=no;"
)
engine = create_engine(f"mssql+pyodbc:///?odbc_connect={quote_plus(conn_str)}")Creating and Using Engines
The engine is a factory for database connections. It manages a connection pool — a set of reusable connections that avoid the overhead of opening a new connection for every query.
from sqlalchemy import create_engine, text
# Create the engine once (typically at application startup)
engine = create_engine(
"postgresql://user:password@localhost:5432/mydb",
pool_size=5, # Keep 5 connections in the pool
max_overflow=10, # Allow up to 10 extra connections under load
pool_timeout=30, # Wait up to 30s for a connection from the pool
pool_recycle=1800, # Recycle connections after 30 minutes (prevents stale connections)
echo=False # Set True to log all SQL statements (debugging)
)
# Execute queries using a connection context manager
with engine.connect() as conn:
result = conn.execute(text("SELECT * FROM orders WHERE amount > :min_amount"),
{"min_amount": 100})
for row in result:
print(row.order_id, row.amount)
# Connection is returned to the pool when the block exits
# Execute with transaction (auto-commit on success, rollback on error)
with engine.begin() as conn:
conn.execute(text("INSERT INTO orders (product, amount) VALUES (:p, :a)"),
{"p": "Widget", "a": 29.99})
conn.execute(text("UPDATE inventory SET qty = qty - 1 WHERE product = :p"),
{"p": "Widget"})
# Both statements commit together, or both roll back on errorSQLAlchemy 2.0 important change: In SQLAlchemy 2.0+, you must wrap raw SQL strings in text(). Passing a plain string like conn.execute("SELECT 1") raises a warning or error. Always use text() for raw SQL.
The text() Function and Parameterized Queries
Never build SQL by concatenating strings — that is the number one cause of SQL injection vulnerabilities. Use text() with named parameters.
from sqlalchemy import text
# WRONG -- SQL injection risk (NEVER do this)
# query = f"SELECT * FROM users WHERE name = '{user_input}'"
# CORRECT -- parameterized query (safe from injection)
with engine.connect() as conn:
result = conn.execute(
text("SELECT * FROM users WHERE name = :name AND role = :role"),
{"name": "Alice", "role": "admin"}
)
rows = result.fetchall()
# Multiple rows with executemany pattern
with engine.begin() as conn:
conn.execute(
text("INSERT INTO users (name, email) VALUES (:name, :email)"),
[
{"name": "Alice", "email": "alice@example.com"},
{"name": "Bob", "email": "bob@example.com"},
{"name": "Charlie", "email": "charlie@example.com"},
]
)pyodbc — Direct SQL Server and Azure SQL Access
pyodbc is the go-to library when you need direct, low-level access to SQL Server, Azure SQL, or Synapse. It bypasses SQLAlchemy and talks directly to the ODBC driver, giving you maximum control over the connection.
Analogy — Calling someone directly on the phone vs going through a receptionist. SQLAlchemy is the receptionist — handles routing, pooling, and protocol translation. pyodbc is a direct phone call — faster, but you handle everything yourself.
Connection Patterns
import pyodbc
# SQL Server (on-premises or VM)
conn = pyodbc.connect(
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=my-server;"
"DATABASE=mydb;"
"UID=sa;"
"PWD=MyPassword123;"
"TrustServerCertificate=yes;"
)
# Azure SQL Database
conn = pyodbc.connect(
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=myserver.database.windows.net;"
"DATABASE=mydb;"
"UID=sqladmin;"
"PWD=MyPassword123;"
"Encrypt=yes;"
"TrustServerCertificate=no;"
"Connection Timeout=30;"
)
# Windows Authentication (Trusted Connection)
conn = pyodbc.connect(
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=my-server;"
"DATABASE=mydb;"
"Trusted_Connection=yes;"
)
# Execute queries
cursor = conn.cursor()
cursor.execute("SELECT TOP 10 * FROM orders WHERE region = ?", ("Ontario",))
rows = cursor.fetchall()
for row in rows:
print(row.order_id, row.amount)
# Always close when done
cursor.close()
conn.close()pyodbc with Context Managers
import pyodbc
# Use a context manager for automatic cleanup
conn_str = (
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=myserver.database.windows.net;"
"DATABASE=mydb;"
"UID=sqladmin;PWD=MyPassword123;"
"Encrypt=yes;TrustServerCertificate=no;"
)
with pyodbc.connect(conn_str) as conn:
cursor = conn.cursor()
# Parameterized query (? placeholders for pyodbc)
cursor.execute(
"SELECT * FROM orders WHERE amount > ? AND region = ?",
(100, "Ontario")
)
rows = cursor.fetchall()
# Get column names from cursor description
columns = [col[0] for col in cursor.description]
print(columns) # ["order_id", "product", "amount", "region"]
# Connection closes automaticallypsycopg2 — Native PostgreSQL Access
psycopg2 is the most popular PostgreSQL adapter for Python. It is faster than going through SQLAlchemy for simple queries because there is no translation layer.
import psycopg2
# Connect to PostgreSQL
conn = psycopg2.connect(
host="localhost",
port=5432,
dbname="mydb",
user="postgres",
password="MyPassword123"
)
# Auto-commit mode (each statement commits immediately)
conn.autocommit = True
# Execute queries
with conn.cursor() as cur:
# Parameterized query (%s placeholders for psycopg2)
cur.execute("SELECT * FROM orders WHERE amount > %s", (100,))
rows = cur.fetchall()
columns = [desc[0] for desc in cur.description]
for row in rows:
print(dict(zip(columns, row)))
conn.close()
# Using RealDictCursor for dict-style access
from psycopg2.extras import RealDictCursor
with psycopg2.connect(host="localhost", dbname="mydb",
user="postgres", password="pass") as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("SELECT * FROM orders LIMIT 5")
for row in cur:
print(row["order_id"], row["amount"]) # Dict-style accesssqlite3 — Built-in Local Database
sqlite3 is built into Python — no installation needed. It creates a single-file database, making it perfect for local development, testing, prototyping, and embedded applications.
Analogy — A personal notebook vs a shared office filing cabinet. PostgreSQL and SQL Server are filing cabinets in a shared office — multiple people access them simultaneously. SQLite is your personal notebook — fast, portable, always available, but meant for one user at a time.
import sqlite3
# Create (or open) a local database file
conn = sqlite3.connect("local.db")
conn.row_factory = sqlite3.Row # Enable dict-style access
cursor = conn.cursor()
# Create a table
cursor.execute("""
CREATE TABLE IF NOT EXISTS orders (
order_id INTEGER PRIMARY KEY AUTOINCREMENT,
product TEXT NOT NULL,
amount REAL NOT NULL,
order_date TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Insert data
cursor.execute("INSERT INTO orders (product, amount) VALUES (?, ?)",
("Widget", 29.99))
conn.commit()
# Bulk insert
orders = [
("Gadget", 49.99),
("Doohickey", 9.99),
("Thingamajig", 74.50),
]
cursor.executemany("INSERT INTO orders (product, amount) VALUES (?, ?)", orders)
conn.commit()
# Query with dict-style access
for row in cursor.execute("SELECT * FROM orders WHERE amount > ?", (20,)):
print(dict(row)) # {"order_id": 1, "product": "Widget", ...}
conn.close()
# In-memory database (vanishes when connection closes -- perfect for tests)
conn = sqlite3.connect(":memory:")pandas read_sql() and to_sql() — DataFrame to Database Bridge
pandas integrates directly with SQLAlchemy and native database connections. read_sql() loads query results into a DataFrame. to_sql() writes a DataFrame to a database table. These two functions are the bridge between your data processing (pandas) and your data storage (database).
Reading from Databases
import pandas as pd
from sqlalchemy import create_engine, text
engine = create_engine("postgresql://user:pass@localhost:5432/mydb")
# Read a full table
df = pd.read_sql("orders", engine)
# Read with a SQL query
df = pd.read_sql("SELECT * FROM orders WHERE amount > 100", engine)
# Read with parameters (SQLAlchemy text() required)
df = pd.read_sql(
text("SELECT * FROM orders WHERE region = :region AND amount > :min"),
engine,
params={"region": "Ontario", "min": 50}
)
# Read with date parsing
df = pd.read_sql("SELECT * FROM orders", engine, parse_dates=["order_date"])
# Read in chunks (large tables)
for chunk in pd.read_sql("SELECT * FROM big_table", engine, chunksize=50_000):
processed = chunk[chunk["status"] == "active"]
# Process each chunk independently
# Read with pyodbc directly (for SQL Server)
import pyodbc
conn = pyodbc.connect("DRIVER={ODBC Driver 18 for SQL Server};SERVER=...;DATABASE=...;UID=...;PWD=...")
df = pd.read_sql("SELECT TOP 1000 * FROM orders", conn)
conn.close()Writing to Databases
import pandas as pd
from sqlalchemy import create_engine
engine = create_engine("postgresql://user:pass@localhost:5432/mydb")
# Write DataFrame to a new table
df.to_sql("orders_staging", engine, index=False, if_exists="replace")
# if_exists options:
# "fail" -- raise error if table exists (default)
# "replace" -- drop and recreate the table
# "append" -- insert rows into existing table
# Write with specific data types
from sqlalchemy import types
df.to_sql("orders", engine, index=False, if_exists="replace",
dtype={
"order_id": types.BigInteger(),
"product": types.String(100),
"amount": types.Numeric(10, 2),
"order_date": types.DateTime()
})
# Write in chunks (large DataFrames)
df.to_sql("big_table", engine, index=False, if_exists="append",
chunksize=10_000, method="multi")
# method="multi" inserts multiple rows per INSERT statement (much faster)
# Write with a custom schema
df.to_sql("orders", engine, index=False, schema="staging", if_exists="replace")Connection Pooling
Opening a database connection is expensive — it involves TCP handshake, authentication, protocol negotiation, and memory allocation on the server. A connection pool keeps a set of open connections ready for reuse, eliminating this overhead for every query.
Analogy — A taxi stand at an airport. Without a pool, every passenger (query) must call a taxi from scratch (open a new connection). With a pool, taxis wait at the stand (idle connections). A passenger grabs one instantly, uses it, and returns it to the stand for the next passenger.
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool, NullPool
# Default pool settings (QueuePool)
engine = create_engine(
"postgresql://user:pass@localhost:5432/mydb",
pool_size=10, # 10 connections kept open
max_overflow=20, # Up to 20 extra under burst load
pool_timeout=30, # Wait 30s for a connection before raising an error
pool_recycle=1800, # Recycle connections every 30 minutes
pool_pre_ping=True # Test connection health before using it
)
# No pool (for scripts that run once and exit)
engine = create_engine(
"postgresql://user:pass@localhost:5432/mydb",
poolclass=NullPool # Every connect() creates a new connection
)
# Check pool status
print(f"Pool size: {engine.pool.size()}")
print(f"Checked out: {engine.pool.checkedout()}")
print(f"Overflow: {engine.pool.overflow()}")Transactions — All or Nothing
A transaction groups multiple SQL statements into one atomic unit. Either all of them succeed (commit) or all of them fail (rollback). This prevents partial updates that leave your database in an inconsistent state.
Analogy — A bank wire transfer. Transferring $500 from Account A to Account B involves two operations: debit A and credit B. If the debit succeeds but the credit fails, $500 vanishes. A transaction ensures both happen together or neither does.
from sqlalchemy import create_engine, text
engine = create_engine("postgresql://user:pass@localhost:5432/mydb")
# engine.begin() auto-commits on success, auto-rolls-back on exception
try:
with engine.begin() as conn:
conn.execute(text("UPDATE accounts SET balance = balance - 500 WHERE id = :a"),
{"a": 1})
conn.execute(text("UPDATE accounts SET balance = balance + 500 WHERE id = :b"),
{"b": 2})
# If we reach here, both updates commit
except Exception as e:
print(f"Transaction rolled back: {e}")
# Both updates are undone
# Manual transaction control with connect()
with engine.connect() as conn:
trans = conn.begin()
try:
conn.execute(text("DELETE FROM orders WHERE status = :s"), {"s": "cancelled"})
conn.execute(text("INSERT INTO audit_log (action) VALUES (:a)"),
{"a": "purged_cancelled_orders"})
trans.commit()
except Exception:
trans.rollback()
raiseStoring Credentials Securely with Environment Variables
Never hardcode database passwords in your scripts. Use environment variables, .env files, or secret managers.
import os
from sqlalchemy import create_engine
# Read credentials from environment variables
db_user = os.environ["DB_USER"]
db_pass = os.environ["DB_PASSWORD"]
db_host = os.environ["DB_HOST"]
db_name = os.environ["DB_NAME"]
engine = create_engine(f"postgresql://{db_user}:{db_pass}@{db_host}:5432/{db_name}")
# Using python-dotenv for .env files
from dotenv import load_dotenv
load_dotenv() # Reads .env file in the current directory
engine = create_engine(
f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
f"@{os.getenv('DB_HOST')}:5432/{os.getenv('DB_NAME')}"
)
# .env file (add to .gitignore -- NEVER commit this)
# DB_USER=postgres
# DB_PASSWORD=MySecretPassword
# DB_HOST=localhost
# DB_NAME=production_dbConnecting to Azure SQL with Managed Identity
In production Azure environments, you should use Managed Identity instead of username/password. This eliminates stored credentials entirely — Azure handles authentication automatically.
import pyodbc
import struct
from azure.identity import DefaultAzureCredential
# Get an access token from Azure AD
credential = DefaultAzureCredential()
token = credential.get_token("https://database.windows.net/.default")
access_token = token.token
# Encode the token for pyodbc
token_bytes = access_token.encode("utf-8")
token_struct = struct.pack("<I", len(token_bytes)) + token_bytes
# Connect without username/password
conn = pyodbc.connect(
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=myserver.database.windows.net;"
"DATABASE=mydb;"
"Encrypt=yes;TrustServerCertificate=no;",
attrs_before={1256: token_struct}
)
cursor = conn.cursor()
cursor.execute("SELECT TOP 5 * FROM orders")
for row in cursor.fetchall():
print(row)
conn.close()Production Pattern: Reusable Database Connection Manager
In production pipelines, you want a centralized connection manager that handles engine creation, connection pooling, health checks, and cleanup. Here is a pattern you can use across all your scripts.
import os
from contextlib import contextmanager
from sqlalchemy import create_engine, text
class DatabaseManager:
# Centralized database connection manager with pooling and health checks.
def __init__(self, connection_url=None):
url = connection_url or os.environ.get("DATABASE_URL")
if not url:
raise ValueError("No connection URL provided")
self.engine = create_engine(
url,
pool_size=5,
max_overflow=10,
pool_recycle=1800,
pool_pre_ping=True
)
def health_check(self):
# Verify the database is reachable.
with self.engine.connect() as conn:
conn.execute(text("SELECT 1"))
return True
@contextmanager
def connection(self):
# Yield a connection that auto-closes on exit.
conn = self.engine.connect()
try:
yield conn
finally:
conn.close()
@contextmanager
def transaction(self):
# Yield a connection with auto-commit/rollback transaction.
with self.engine.begin() as conn:
yield conn
def read_df(self, query, params=None):
# Execute a query and return a pandas DataFrame.
import pandas as pd
return pd.read_sql(text(query), self.engine, params=params)
def write_df(self, df, table_name, if_exists="append", schema=None):
# Write a DataFrame to a database table.
df.to_sql(table_name, self.engine, index=False,
if_exists=if_exists, schema=schema,
chunksize=10_000, method="multi")
def close(self):
self.engine.dispose()
# Usage
db = DatabaseManager("postgresql://user:pass@localhost:5432/mydb")
# Health check
db.health_check()
# Read into DataFrame
df = db.read_df("SELECT * FROM orders WHERE region = :r", {"r": "Ontario"})
# Write DataFrame
db.write_df(df, "orders_backup")
# Transaction
with db.transaction() as conn:
conn.execute(text("DELETE FROM staging_orders"))
conn.execute(text("INSERT INTO staging_orders SELECT * FROM raw_orders"))
db.close()Common Mistakes
1. Hardcoding database passwords in scripts. Credentials in source code end up in Git history, shared repos, and logs. Use environment variables, .env files (with .gitignore), or cloud secret managers (Azure Key Vault, AWS Secrets Manager). Never commit passwords.
2. Not using parameterized queries. Building SQL by string concatenation (f"SELECT * FROM users WHERE name = '{name}'") is vulnerable to SQL injection. Always use text() with :named parameters in SQLAlchemy, ? in pyodbc, or %s in psycopg2.
3. Forgetting to close connections. Every unclosed connection stays open on the database server, consuming memory and counting against the connection limit. Use context managers (with engine.connect() or with pyodbc.connect()) to ensure connections always close.
4. Creating a new engine for every query. create_engine() sets up a connection pool — it should be called once at application startup, not inside a loop or function. Creating an engine per query creates a new pool each time, defeating the purpose.
5. Using if_exists="replace" in production. to_sql(if_exists="replace") drops and recreates the table, losing indexes, constraints, permissions, and any data already there. In production, use "append" and manage schema separately.
6. Passing raw strings to conn.execute() in SQLAlchemy 2.0+. SQLAlchemy 2.0 requires text() for raw SQL. conn.execute("SELECT 1") raises RemovedIn20Warning or errors. Always wrap raw SQL in text().
7. Not setting pool_recycle for long-running applications. Database servers close idle connections after a timeout (typically 30 minutes). If your pool holds stale connections, queries fail with “connection reset” errors. Set pool_recycle=1800 and pool_pre_ping=True.
8. Using sqlite3 in production with concurrent users. SQLite locks the entire database on writes — only one writer at a time. Multiple concurrent users will get “database is locked” errors. Use PostgreSQL or SQL Server for multi-user applications.
Interview Questions
Q: What is SQLAlchemy and why do data engineers use it instead of native database drivers? A: SQLAlchemy is a Python SQL toolkit that provides a universal interface to any relational database. Data engineers use it because one codebase works with SQL Server, PostgreSQL, MySQL, and SQLite — only the connection string changes. It also provides connection pooling (reuses connections instead of opening new ones), transaction management (commit/rollback), parameterized queries (prevents SQL injection), and integration with pandas (read_sql/to_sql). Native drivers like pyodbc or psycopg2 are faster for database-specific operations but require rewriting code when switching databases.
Q: What is a connection pool and why does it matter? A: A connection pool maintains a set of open database connections that are reused across queries. Opening a new connection is expensive (TCP handshake, authentication, memory allocation). With a pool, the first query opens a connection, and subsequent queries reuse it instead of opening new ones. This reduces latency from hundreds of milliseconds to near-zero for pooled connections. Key parameters are pool_size (idle connections kept open), max_overflow (extra connections under load), pool_recycle (how often to refresh connections), and pool_pre_ping (test connection health before use).
Q: How do you prevent SQL injection in Python? A: Use parameterized queries — never concatenate user input into SQL strings. In SQLAlchemy, use text() with named parameters: text("SELECT <em> FROM users WHERE name = :name") with {"name": user_input}. In pyodbc, use ? placeholders: cursor.execute("SELECT </em> FROM users WHERE name = ?", (user_input,)). In psycopg2, use %s: cur.execute("SELECT * FROM users WHERE name = %s", (user_input,)). The database driver handles escaping and quoting, making injection impossible.
Q: What is the difference between engine.connect() and engine.begin() in SQLAlchemy? A: engine.connect() gives you a connection where you manage transactions manually — you must call conn.commit() or conn.rollback() yourself. engine.begin() gives you a connection wrapped in an automatic transaction — it commits when the block exits normally and rolls back if an exception occurs. Use begin() for write operations (INSERT, UPDATE, DELETE) that need atomicity. Use connect() for read-only queries or when you need manual transaction control.
Q: How would you load a 10 GB database table into a pandas DataFrame? A: You would not load 10 GB at once. Instead, use chunksize in read_sql() to stream the data in batches: for chunk in pd.read_sql("SELECT * FROM big_table", engine, chunksize=50_000). Process each 50,000-row chunk independently. Alternatively, push the filtering to the database with a WHERE clause so you only read the subset you need. For truly large tables, consider switching to PySpark which distributes the load across a cluster.
Q: What is pool_pre_ping and when should you use it? A: pool_pre_ping=True tells SQLAlchemy to send a lightweight test query (like SELECT 1) before handing a connection to your code. If the test fails (connection dropped by the database), the pool discards it and opens a fresh one. Use it in long-running applications (web servers, scheduled pipelines) where connections may go stale between uses. The overhead is minimal (one round-trip per checkout) compared to the cost of a failed query on a dead connection.
Q: When would you use pyodbc directly instead of SQLAlchemy? A: Use pyodbc directly when you need SQL Server-specific features that SQLAlchemy does not expose, such as Azure AD token-based authentication (passing token structs via attrs_before), bulk copy operations (fast_executemany), or direct ODBC driver configuration. Also use pyodbc for simple scripts that connect to one SQL Server database, run one query, and exit — the SQLAlchemy overhead (engine setup, pool management) is unnecessary for one-shot scripts.
Wrapping Up
Database connections are the plumbing of data engineering — invisible when working, catastrophic when broken. Master SQLAlchemy for portable, production-grade connections. Know pyodbc for SQL Server and Azure SQL specifics. Use psycopg2 for PostgreSQL performance. Keep sqlite3 in your toolkit for local development and testing. And always, always, always: parameterize your queries, pool your connections, and keep your passwords out of your code.
Related posts: — File Formats (CSV, JSON, Parquet, Excel) — Working with APIs & HTTP — Python for Data Engineers — Azure SQL Database Guide
Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.