Stop Wiring Up Database Drivers Manually — A Simpler Python Database API

Adebayo Olaonipekun — Fri, 05 Jun 2026 16:54:46 +0000

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", "alice@example.com")
    )

    # Fetch all rows
    db.execute_query("SELECT * FROM users")
    rows = db.fetch_all()
    print(rows)  # [(1, 'Alice', 'alice@example.com')]

    # Fetch by parameter
    result = db.fetch_by_param("users", "name", "Alice")
    print(result)  # [(1, 'Alice', 'alice@example.com')]

Bulk inserts

users = [
    {"name": "Bob", "email": "bob@example.com"},
    {"name": "Charlie", "email": "charlie@example.com"},
    {"name": "David", "email": "david@example.com"},
]

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.

DEV Community: Adebayo Olaonipekun