py-store

MongoDB → PostgreSQL migration

The problem

Your service started on MongoDB and now needs to run on PostgreSQL — in production, per customer, or on both at once during a transition. Rewriting every data-access call site into SQL is expensive, and keeping two query layers in sync is worse.

Why py-store

Walkthrough

from pymongo import AsyncMongoClient
from py_store import init, store

# The schema is backend-agnostic: register it once, then point init() at any backend.
store.register({
    "name": "OrderItem",
    "collection": "order_items",
    "fields": {
        "_id": "string",
        "orderId": {"type": "string", "default": ""},
        "sku": {"type": "string", "default": ""},
        "qty": {"type": "int", "default": 1},
    },
})

order_schema = {
    "name": "Order",
    "collection": "orders",
    "idPrefix": "OD",
    "fields": {
        "_id": "string",
        "customerId": {"type": "string", "default": ""},
        "amount": {"type": "float", "default": 0},
        "status": {"type": "string", "default": "open"},
    },
    "relations": {
        "items": {"model": "OrderItem", "type": "many",
                  "localField": "_id", "foreignField": "orderId"},
    },
    "computes": {
        "itemCount": {"type": "int", "agg": {"$count": "items"}},
    },
    "indexes": [{"keys": {"status": 1, "createdAt": -1}}],
}
store.register(order_schema)

GQL = "Order($condition:@c0,$sort:@s1,$limit:@l) { _id, customerId, amount, itemCount }"
PARAMS = {"c0": {"status": "open"}, "s1": {"createdAt": -1}, "l": 20}

# --- Phase 1: MongoDB (native aggregation + $lookup) ---
mongo = AsyncMongoClient("mongodb://localhost:27017")
await init(mongo["mydb"])
mongo_rows = await store.query(GQL, PARAMS)

# --- Phase 2: PostgreSQL — same schema, same GQL, only the datasource changes ---
await init({"default": {"kind": "postgres", "exec": exec}})
pg_rows = await store.query(GQL, PARAMS)          # parameterized SQL + JOIN

# Optionally align against an existing physical database first.
# sync_schema only READS structure: introspect -> merge overlay -> register.
# Signature: sync_schema(backend, driver, introspect_options=None, overlay=None,
#                        datasource=None, namespace=None, register_defs=True)
defs = await store.sync_schema(
    "postgres",
    driver,                  # prefer a read-only account
    overlay=[order_schema],  # local schemaJSON merged on top (permissions / computes)
)

Pitfalls

See also