If you've ever had to support more than one database in a Python project, you know the drill.
SQLite in development, PostgreSQL in production. A client insists on MySQL. A legacy system runs on SQL Server. Your ML pipeline writes to Oracle. Each one needs its own driver, its own connection string format, its own placeholder syntax (? vs %s vs $1), and its own quirks around transactions and pooling.
You end up with five different code paths for what is conceptually the same operation: connect, query, commit, close.
This post walks through a pattern for unifying that — and introduces a small open-source library I built to handle the boilerplate.
The Problem in Code
Here's what connecting to three different databases typically looks like:
# SQLite
import sqlite3
conn = sqlite3.connect("local.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE name = ?", ("Alice",))
# PostgreSQL
import psycopg2
conn = psycopg2.connect(host="localhost", user="user", password="pass", dbname="mydb")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE name = %s", ("Alice",))
# MySQL
import mysql.connector
conn = mysql.connector.connect(host="localhost", user="user", password="pass", database="mydb")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE name = %s", ("Alice",))
Three drivers, three import styles, two different placeholder characters, three different connection parameter names (dbname vs database). And we haven't even touched connection pooling, transactions, or error handling yet.
A Unified API
What if connecting to any of those looked like this instead:
from sqlpyhelper.db_helper import SQLPyHelper
# SQLite
with SQLPyHelper(db_type="sqlite", database="local.db") as db:
result = db.fetch_by_param("users", "name", "Alice")
# PostgreSQL
with SQLPyHelper(db_type="postgres", host="localhost",
user="user", password="pass", database="mydb") as db:
result = db.fetch_by_param("users", "name", "Alice")
# MySQL
with SQLPyHelper(db_type="mysql", host="localhost",
user="user", password="pass", database="mydb") as db:
result = db.fetch_by_param("users", "name", "Alice")
Same method, same result, regardless of the database underneath. The context manager handles connection cleanup automatically — no manual conn.close() needed.
Getting Started
pip install sqlpyhelper # SQLite works out of the box
pip install "sqlpyhelper[postgres]" # add PostgreSQL
pip install "sqlpyhelper[mysql]" # add MySQL
pip install "sqlpyhelper[sqlserver]" # add SQL Server
pip install "sqlpyhelper[oracle]" # add Oracle
pip install "sqlpyhelper[all]" # everything
Only the drivers you actually need get installed. If you're building a SQLite-only tool you don't get cx_Oracle forced on you.
Common Operations
Querying
from sqlpyhelper.db_helper import SQLPyHelper
with SQLPyHelper(db_type="sqlite", database="myapp.db") as db:
# Create a table
db.create_table("users", {
"id": "INTEGER PRIMARY KEY",
"name": "TEXT",
"email": "TEXT"
})
# Insert a row
db.execute_query(
"INSERT INTO users (name, email) VALUES (?, ?)",
("Alice", "[email protected]")
)
# Fetch all rows
db.execute_query("SELECT * FROM users")
rows = db.fetch_all()
print(rows) # [(1, 'Alice', '[email protected]')]
# Fetch by parameter
result = db.fetch_by_param("users", "name", "Alice")
print(result) # [(1, 'Alice', '[email protected]')]
Bulk inserts
users = [
{"name": "Bob", "email": "[email protected]"},
{"name": "Charlie", "email": "[email protected]"},
{"name": "David", "email": "[email protected]"},
]
with SQLPyHelper(db_type="postgres", host="localhost",
user="user", password="pass", database="mydb") as db:
db.insert_bulk("users", users)
Uses executemany under the hood — efficient for large datasets.
Transactions
with SQLPyHelper(db_type="postgres", host="localhost",
user="user", password="pass", database="mydb") as db:
db.begin_transaction()
try:
db.execute_query("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
db.execute_query("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))
db.commit_transaction()
except Exception:
db.rollback_transaction()
raise
Connection pooling
with SQLPyHelper(db_type="postgres", host="localhost",
user="user", password="pass", database="mydb") as db:
db.setup_connection_pool(min_conn=2, max_conn=10)
conn = db.get_connection_from_pool()
# use conn
db.return_connection_to_pool(conn)
Async Support for FastAPI
If you're building a FastAPI application, there's an async-native class that uses asyncpg, aiomysql, and aiosqlite under the hood:
pip install "sqlpyhelper[async-postgres]"
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlpyhelper.async_helper import AsyncSQLPyHelper
db = AsyncSQLPyHelper(
db_type="postgres",
host="localhost",
user="user",
password="pass",
database="mydb",
)
@asynccontextmanager
async def lifespan(app: FastAPI):
await db.connect()
yield
await db.close()
app = FastAPI(lifespan=lifespan)
@app.get("/users")
async def get_users():
return await db.fetch_all("SELECT * FROM users")
@app.post("/users")
async def create_user(name: str):
await db.execute(
"INSERT INTO users (name) VALUES ($1)", name
)
return {"status": "created"}
Notice the $1 placeholder — the async class uses a unified placeholder syntax across all databases and translates automatically to ?, %s, or :1 depending on the driver. You write it once and it works everywhere.
Cross-Database Migration
One feature that doesn't exist cleanly anywhere else: copying a table from one database type to another.
from sqlpyhelper.db_helper import SQLPyHelper
from sqlpyhelper.migration import migrate_table
# Promote a SQLite dev database to PostgreSQL production
with SQLPyHelper(db_type="sqlite", database="dev.db") as source:
with SQLPyHelper(db_type="postgres", host="prod-host",
user="user", password="pass", database="prod_db") as target:
stats = migrate_table(
source=source,
target=target,
table="users",
create_table=True, # auto-creates with type mapping
batch_size=500,
)
print(f"Migrated {stats['rows_migrated']} rows")
migrate_table inspects the source schema, maps column types to the target database's equivalents (e.g. SQLite TEXT → PostgreSQL TEXT, SQLite REAL → PostgreSQL DOUBLE PRECISION), creates the table, and inserts in batches.
It returns a stats dict so you can log or verify:
{
"table": "users",
"rows_migrated": 1234,
"batches": 3,
"source_db": "sqlite",
"target_db": "postgres"
}
Error Handling
Rather than silently returning None on failure, the library raises typed exceptions:
from sqlpyhelper.db_helper import SQLPyHelper, QueryError, ConnectionError
try:
with SQLPyHelper(db_type="sqlite", database="myapp.db") as db:
db.execute_query("SELECT * FROM nonexistent_table")
except QueryError as e:
print(f"Query failed: {e}")
except ConnectionError as e:
print(f"Connection failed: {e}")
This makes it straightforward to handle database errors in application code without catching bare Exception.
Environment Variables
For production use, credentials can be loaded from a .env file:
# .env
DB_TYPE=postgres
DB_HOST=localhost
DB_USER=myuser
DB_PASSWORD=mypassword
DB_NAME=mydb
from sqlpyhelper.db_helper import SQLPyHelper
db = SQLPyHelper() # reads from .env automatically
When NOT to use this
To be honest about the tradeoffs:
Use SQLAlchemy instead if you:
- Need ORM capabilities (mapping Python classes to tables)
- Are building a complex application with many relationships and migrations
- Need advanced query building via Python expressions
- Already have a SQLAlchemy codebase
SQLPyHelper is a better fit when you:
- Need raw SQL with a clean connection API
- Are building scripts, data pipelines, or utilities that touch multiple databases
- Want to avoid ORM overhead for straightforward CRUD operations
- Need to migrate data between database types
Links
- PyPI: pypi.org/project/SQLPyHelper
- GitHub: github.com/adebayopeter/sqlpyhelper
- Documentation: sqlpyhelper.readthedocs.io
The library is MIT licensed and open to contributions. If you try it and hit something that doesn't work the way you'd expect, open an issue — the goal is to make it genuinely useful, not just theoretically complete.