What is pyflashdb?

pyflashdb lets you talk to MySQL, PostgreSQL, SQLite and MongoDB with one API — and the same filters, ordering, null handling, upserts and transactions give the same results on each of them. No SQL required, no driver switching, no boilerplate.

from pyflashdb import FlashDB

with FlashDB.from_url("sqlite:///app.db") as flash:   # or postgresql://…, mysql://…, mongodb://…
    flash.add("users", {"name": "Alice", "age": 25})                  # insert
    flash.where("users", {"age": {">": 18}})                          # filter
    flash.update("users", {"name": "Alice"}, {"age": 26})             # update
    flash.upsert("users", {"email": "a@x.com", "name": "Alice"}, key="email")
    flash.delete("users", {"name": "Alice"})                          # delete

Change the URL and nothing else changes.

The claim is tested. tests/test_parity.py runs one set of assertions against real MySQL, PostgreSQL, MongoDB and SQLite in CI. If any backend returns different rows, that backend's run fails.

Install

pip install pyflashdb              # SQLite works out of the box — no driver needed
pip install "pyflashdb[mysql]"     # + MySQL      (mysql-connector-python)
pip install "pyflashdb[postgres]"  # + PostgreSQL (psycopg2-binary)
pip install "pyflashdb[mongodb]"   # + MongoDB    (pymongo)
pip install "pyflashdb[all]"       # all three drivers
pip install "pyflashdb[pandas]"    # + flash.to_df()

The package is pyflashdb and it is imported as pyflashdb. (The old import flash still works in 2.x but is deprecated — see Upgrading.)

Connect

import os
from pyflashdb import FlashDB

flash = FlashDB.from_url(os.environ["DATABASE_URL"])     # passwords stay out of your code
flash = FlashDB.from_env()                               # same, reads $DATABASE_URL
flash = FlashDB.from_url("postgresql://user:pass@localhost:5432/app", echo=True)   # print every query

flash = FlashDB("mysql", {"host": "localhost", "user": "root", "password": "…", "database": "app"})

Options (as URL query string or config keys): pool_size (connections, default 1), pool_timeout, retries, retry_delay, ping_after, connect_args (passed to the driver: ssl, …). A pre-built pymongo.MongoClient can be supplied as config["client"].

Always use it as a context manager (or call flash.close()):

with FlashDB.from_url(url) as flash:
    ...

Everyday API

Filters — write once, same result everywhere

flash.where("users", {"age": {">=": 18, "<=": 65}})
flash.where("users", {"role": {"in": ["admin", "mod"]}})
flash.where("users", {"status": {"!=": "banned"}})
flash.where("users", {"email": {"like":  "%@gmail.com"}})     # case-SENSITIVE
flash.where("users", {"email": {"ilike": "%@gmail.com"}})     # case-insensitive
flash.where("users", {"deleted_at": None})                    # IS NULL / missing
flash.select("users", fields=["name", "age"], order_by=["-age", "name"], limit=10, offset=20)

These rules hold on every backend (they are what the parity suite checks):

• like is case-sensitive and anchored: % = any run of characters, _ = one character, \% / \_ = literal. Every other character is literal, so the . in %@gmail.com is a dot. ilike is the case-insensitive form.
• None matches NULL/missing. != and not in include NULLs (plain SQL would drop them). in [] matches nothing; not in [] matches everything.
• order_by="-age" sorts descending; NULLs sort first ascending, last descending.
• update() returns rows matched, even if a value did not change.
• Unknown operators raise FilterError — nothing user-supplied is ever pasted into SQL, and every table/column name is quoted.

Return types

The returned id can be used straight in a filter on every backend: flash.find_one("users", {"id": new_id}) (SQL) / {"_id": new_id} (Mongo — a 24-hex string is treated as an ObjectId). Values come back as native Python types everywhere: bool, datetime (naive UTC), Decimal, and dict/list for json columns.

Upsert

flash.upsert("users", {"email": "a@x.com", "name": "Alice"}, key="email")
flash.upsert("users", data, key=["org_id", "email"], update_fields=["name"])   # only overwrite `name`
flash.upsert_many("users", rows, key="email")

On SQL the key must be backed by a primary key or unique index ({"unique": True} in create_table). On MySQL the update fires on any unique-key conflict.

Transactions

with flash.transaction():                       # commits on success, rolls back on any exception
    flash.update("accounts", {"user_id": 1}, {"balance": 400})
    flash.update("accounts", {"user_id": 2}, {"balance": 600})

begin() / commit() / rollback() still exist. A transaction() inside another one joins it. Transactions are per thread.

MongoDB needs a replica set. On a standalone mongod (what most people run locally) begin() raises a TransactionError explaining how to fix it, instead of failing confusingly. For local development run a single-node replica set — docker compose up -d in this repository does that — and connect with ?replicaSet=rs0. Atlas works out of the box.

(MySQL DDL such as TRUNCATE / ALTER TABLE commits implicitly, as in MySQL itself.)

Hooks & plugins

Before/after hooks for every data operation — also on MongoDB, which has no triggers:

@flash.before_insert("users")
def validate(data):                 # a dict for add(), a list of dicts for bulk_insert()
    if not data.get("email"):
        raise ValueError("email required")      # aborts the insert

@flash.after_insert("users")
def on_created(data, result):
    print("new id:", result)

@flash.before_select("orders")      # may rewrite the query — e.g. multi-tenant scoping
def scope(payload):
    payload["filters"] = {**(payload["filters"] or {}), "tenant_id": 7}

@flash.before_query                 # sees every statement sent to the database
def log_sql(info):
    print(info["backend"], info["statement"], info["params"])

Events: insert, update, delete, select, upsert, count — each with before_* / after_* ("*" hooks every table) — plus before_query / after_query. before hooks can modify the payload and abort by raising; an exception in an after hook is logged, not raised.

Built-in plugins:

from pyflashdb.plugins import Timestamps, AuditLog

Timestamps(["users", "orders"]).install(flash)   # created_at / updated_at maintained for you
AuditLog(["users"]).install(flash)               # logs every write to the `pyflashdb.audit` logger

Schema

flash.create_table("users", {
    "id":      "int",                                   # auto-increment primary key
    "email":   {"type": "str", "unique": True, "nullable": False},
    "name":    "str(100)",                              # VARCHAR(100)
    "score":   {"type": "float", "default": 0},
    "balance": "decimal(10,2)",
    "active":  {"type": "bool", "default": True},
    "meta":    "json",
    "joined":  "datetime",
})
flash.add_column("users", "nickname", {"type": "str", "nullable": True})
flash.create_index("users", ["name", "joined"])         # idempotent; unique=True for a unique index
flash.table_exists("users")

Types: int bigint smallint str text float double decimal bool date datetime timestamp json blob. Spec keys: type length precision scale primary_key autoincrement unique nullable default collation. An unrecognised type is passed through only if it looks like a plain type name (BIGINT, VARCHAR(50)); anything else raises SchemaError.

On MongoDB, create_table applies a $jsonSchema validator (types, nullable: False → required) and unique indexes; server-side defaults don't exist there and are ignored with a warning.

Large tables, export and copying between databases

for batch in flash.iter("orders", batch_size=1000):     # server-side cursor — never loads it all
    process(batch)

flash.to_csv("users", "users.csv", fields=["name", "email"])
flash.to_json("users", "users.json", indent=2)           # lines=True for JSON Lines
df = flash.to_df("users", filters={"age": {">": 18}})    # pip install "pyflashdb[pandas]"

Copy between databases — "MySQL locally, PostgreSQL in production", made literal:

local = FlashDB.from_url("mysql://root:pw@localhost/app")
prod  = FlashDB.from_url(os.environ["PROD_DATABASE_URL"])

local.copy_to(prod, tables=["users", "orders"], batch_size=1000)
# {'users': 1200, 'orders': 8431}
local.copy_to(prod, tables=["users"], mode="upsert", key="email")      # or mode="replace"

Missing destination tables are created from the source's columns, types, nullability and primary key (indexes, defaults and foreign keys are not copied). Ids are copied as-is between SQL databases (PostgreSQL counters are re-synced). Between SQL and MongoDB each side keeps its own key.

Production

• Pooling — pool_size=10 (MySQL, PostgreSQL). One shared pool implementation; MongoDB's driver pools by itself. The default (1) is one connection shared safely across threads.
• Reconnect & retry — idle connections are health-checked, dead ones replaced. If a connection drops mid-read the read is retried (retries=2). Writes are never retried blindly, because the server may already have applied them; the connection is replaced so the next call works.
• Observability — echo=True, before_query / after_query hooks (with duration and row count), and the pyflashdb.query / pyflashdb.audit / pyflashdb.hooks loggers.
• Typed — full type hints and a py.typed marker, so IDE completion and mypy work.
• raw() — flash.raw("SELECT * FROM users WHERE age > %s", [18]). The placeholder is %s on every SQL backend, SQLite included (write %% for a literal percent sign when passing params). Statements that return rows give a list[dict] (possibly empty); others give the row count. MongoDB: flash.raw({"age": {"$gt": 18}}, table="users") or flash.raw("dbStats").

Known differences

pyflashdb removes the differences that can be removed. These remain, by nature of the databases:

• Ids: int vs str (above). MongoDB documents contain only the fields you stored — a missing field is not turned into None.
• String collation: tables created by create_table use a binary collation on MySQL, so =, IN and LIKE are case-sensitive like everywhere else. On tables you created yourself, = follows the column's own collation (like / ilike are correct regardless). Sort order of non-ASCII text and non-ASCII case folding for ilike follow each database's locale.
• SQLite stores decimal as a numeric value (precision beyond ~15 digits is not guaranteed) and has a single writer, so it uses one connection.
• MongoDB: transactions need a replica set; bulk_insert is not atomic outside a transaction; no column defaults.
• MySQL: upsert fires on any unique key; JSON columns need MySQL ≥ 5.7.

Upgrading from 1.x

2.0 is a major release. What to change:

• Import pyflashdb: from pyflashdb import FlashDB. import flash still works for now but warns and will be removed in a later release.
• Python ≥ 3.9 (3.8 is end-of-life).
• before_* hooks can now abort (raise to cancel the operation) — in 1.x exceptions were swallowed and printed. after_* hook errors are logged via logging, no longer printed.
• describe() returns [{"name", "type", "nullable", "default", "primary_key"}] on every backend.
• != / not in now include NULLs on SQL (as MongoDB always did); like is case-sensitive on MySQL (it was case-insensitive); order_by="-x" now really sorts descending on SQL; update() returns rows matched (MySQL used to return rows changed); Mongo select(fields=…) no longer includes _id.
• Unknown filter operators (including raw $gt-style operators on Mongo) raise FilterError.
• float columns are now double precision (DOUBLE / DOUBLE PRECISION), and timestamp columns on MySQL are DATETIME(6).
• PostgreSQL truncate() no longer cascades into other tables.
• create_index() now works on every backend (it was MongoDB-only).

Full list: Changelog.

How pyflashdb compares

Development

pip install -e ".[all,dev]"
pytest                       # SQLite + unit tests; MySQL/PostgreSQL/Mongo are skipped unless configured
docker compose up -d --wait  # start MySQL, PostgreSQL and a MongoDB replica set, then export:
export PYFLASHDB_TEST_MYSQL_URL=mysql://root:flash@127.0.0.1:3306/flash_test
export PYFLASHDB_TEST_POSTGRES_URL=postgresql://postgres:flash@127.0.0.1:5432/flash_test
export PYFLASHDB_TEST_MONGODB_URL='mongodb://127.0.0.1:27017/flash_test?replicaSet=rs0'
pytest -ra                   # the full parity suite

Documentation

📄 Full Documentation (PDF, v2.0.0) — a 39-page manual covering every method, the cross-database behavior contract, production use, upgrading from 1.x, and the full list of what's new in 2.0 (last chapter).

Links

• PyPI: https://pypi.org/project/pyflashdb/
• Issues: https://github.com/AmanDev17/pyflashdb/issues
• Changelog: https://github.com/AmanDev17/pyflashdb/blob/main/Changelog.md

Made possible because of Komal