py-store

py-store

One data layer for MongoDB, MySQL, SQLite, PostgreSQL and local disk in Python asyncio — define models as pure JSON, query them with a MongoDB-style GQL tree syntax, and get role-based access control, computed columns and soft-delete out of the box.

PyPI version license python versions backends query dialect

py-store lets a Python service talk to MongoDB (native aggregation), MySQL, PostgreSQL, SQLite and local disk through a single schema definition and a single query dialect. Nested relations compile to one native query per backend — you never hand-write $lookup or raw SQL.

Also looking for the Node.js version? See nodejs-store (npm nodejs-store). Both are thin hosts over the shared Rust engine rust-store.

Store family on Gitee: rust-store (shared Rust core) · nodejs-store (Node.js host, npm nodejs-store) · py-store (this repo — Python host, pip storepy)

中文文档见 README.zh-CN.md。

Documentation site: https://coenddt.github.io/py-store/ — every scenario walkthrough with runnable code and the engine’s exact limits, one indexable page per scenario.

Install: the distribution name is storepy; the import package is py_store.

pip install storepy

from py_store import init, store

Table of contents

What it is

A lightweight, backend-agnostic data layer for Python asyncio. You describe your models once as pure JSON (fields, relations, computes, indexes, read/write role whitelists). From that description the library derives:

MongoDB is the primary dialect: queries are written in a MongoDB-flavoured GQL, and the three relational backends adapt to it. That is what makes one schema portable across a document store and three relational stores.

How it relates to nodejs-store and rust-store

                 ┌──────────────────────────────┐
   Node.js  ──▶  │  nodejs-store (npm, host)    │ ─┐
                 └──────────────────────────────┘  │  rust-store-node (napi-rs)
                                                   ▼
                                     ┌───────────────────────────────┐
                                     │ rust-store/core (pure logic)  │
                                     │ GQL · permissions · computes  │
                                     │ command planning · dialects   │
                                     └───────────────────────────────┘
                                                   ▲
                 ┌──────────────────────────────┐  │  rust-store-py (PyO3)
   Python   ──▶  │  py-store (pip, host)        │ ─┘
                 └──────────────────────────────┘

The Rust core owns GQL parsing, permission checks, computed columns, command planning and SQL dialect translation — it never touches a database. The hosts (py-store, nodejs-store) own driver IO, callbacks and placeholder substitution. Behaviour therefore cannot drift between Python and Node.js: there is only one implementation.

When to use it

Reach for py-store when any of these describe your situation:

Typical concrete scenarios (see doc/use-cases/ for full walkthroughs):

Scenario Why py-store fits
Multi-tenant SaaS with per-tenant schema/database database / schema per tenant + runtime route override, one schema
FastAPI / admin backend Schema-driven CRUD, soft-delete, computed columns, RBAC
MongoDB today, PostgreSQL tomorrow Same GQL + same schema, only the datasource changes
AI data-QA / text-to-query agent Plan-only build_pipeline, deterministic command JSON, feedback events
Mixed SQL + Mongo in one product Cross-source queries with native SQL pushdown and Mongo in-memory federation
Audit-friendly CRUD Every schema auto-gets a <Model>Deleted archive table/collection

When not to use it

Being explicit about the boundary saves you time:

How it compares

General positioning, not a benchmark — always verify against each tool’s current docs.

  py-store SQLAlchemy Beanie / Motor Tortoise ORM SQLModel Django ORM
Primary shape JSON schema + GQL data layer SQL toolkit + ORM Async MongoDB ODM / driver Async ORM Pydantic + SQLAlchemy ORM bundled with Django
Backends MongoDB, MySQL, SQLite, PostgreSQL, local PostgreSQL, MySQL, SQLite, Oracle, MSSQL MongoDB PostgreSQL, MySQL, SQLite, Oracle, MSSQL PostgreSQL, MySQL, SQLite, … PostgreSQL, MySQL, SQLite, Oracle
One query dialect across Mongo and SQL ✅ (MongoDB-flavoured GQL) ➖ (SQL only) ➖ (Mongo only) ➖ (SQL only) ➖ (SQL only) ➖ (SQL only)
Nested relation reads in one query ✅ declarative relations → $lookup / JOIN ⚠️ manual selectinload/joins ✅ Link/fetch_links ✅ prefetch_related ⚠️ via SQLAlchemy ✅ prefetch_related
Built-in role / field-level RBAC + owner injection ✅ ➖ ➖ ➖ ➖ ➖ (permissions are app-level)
Read-time computed columns (sync / async / relation-agg) ✅ ➖ (hybrid properties) ➖ ➖ ➖ ➖
Soft-delete archive table auto-provisioned ✅ ➖ ➖ ➖ ➖ ➖
Migration / DDL engine ➖ (introspection read-only; whitelist migrate() orchestrator; optional generate_ddl text) ✅ (Alembic) ➖ ✅ (Aerich) ✅ (Alembic) ✅
Framework coupling none (asyncio) none none none none Django
Shared native core across Python & Node ✅ (Rust rust-store) ➖ ➖ ➖ ➖ ➖

How it differs from specific libraries

Positioning only, based on those projects’ public documentation at the time of writing — verify against your own requirements.

Short version: use an ORM when you want model classes, Pydantic validation and migrations; use py-store when you want one runtime schema + one query dialect spanning MongoDB and SQL, with RBAC and computed columns built in.

Installation

pip install storepy

The distribution name is storepy; the import package is py_store: from py_store import init, store.

Requires Python 3.10+ and one supported backend (MongoDB / MySQL / SQLite / PostgreSQL / local disk).

Optional driver extras:

pip install "storepy[mysql]"     # asyncmy
pip install "storepy[postgres]"  # asyncpg
pip install "storepy[sqlite]"    # aiosqlite

Quick start

from pymongo import AsyncMongoClient
from py_store import init, store

client = AsyncMongoClient("mongodb://localhost:27017")
await init(client["mydb"])   # idempotently creates indexes for registered schemas

# Register a schema (pure JSON)
store.register({
    "name": "Post",                  # model name used in GQL
    "collection": "posts",           # optional, defaults to name
    "idPrefix": "PT",                # string _id: prefix + base36 timestamp + random
    "fields": {
        "title": {"type": "string", "default": ""},
        "status": {"type": "string", "default": "draft"},
        "tags": {"type": "array", "default": []},
    },
    "computes": {
        "statusLabel": {"type": "string", "depends": ["status"],
                        "fn": lambda doc: doc["status"].upper()},
    },
    "indexes": [{"keys": {"status": 1, "createdAt": -1}}],
})

# Write — only user data; defaults are filled on read
doc = await store.insert("Post", {"title": "Hello"})

# Query — GQL tree syntax, values referenced from params via @key
items = await store.query(
    "Post($condition:@c0,$sort:@s1,$limit:@l) { title, status, statusLabel }",
    {"c0": {"status": "draft"}, "s1": {"createdAt": -1}, "l": 20},
)

The same schema and the same query run unchanged against PostgreSQL — only the init() datasource changes:

await init({"default": {"kind": "postgres", "exec": exec}})
items = await store.query("Post($condition:@c0) { title, status }", {"c0": {"status": "draft"}})

Supported backends

Backend Notes
MongoDB native aggregation pipeline (find/aggregate/$lookup), PyMongo AsyncMongoClient (pymongo >= 4.9)
MySQL parameterized SQL, information_schema introspection (asyncmy)
SQLite parameterized SQL, sqlite_master + PRAGMA introspection (aiosqlite)
PostgreSQL parameterized SQL ($n), RETURNING support (asyncpg)
local (local disk) collections persisted as JSON files, evaluated directly by the core’s local evaluator (no SQL, no driver); see Local disk data source (local)

GQL tree queries compile to a single native query per backend (except the local source — commands are evaluated directly by the core’s local evaluator) — never hand-write $lookup or raw SQL again.

Local disk data source (local)

Zero external services, zero native DB engine: collections are persisted as JSON files and commands are evaluated directly inside the Rust core’s local evaluator (semantics match the MongoDB driver).

from py_store import init, local

await init({'default': local.connect({'dir': './data/store'})})

Features

GQL syntax

Model($condition:@c0,$sort:@s1,$skip:@sk,$limit:@l1) {
  field1, field2, obj.subField,
  Relation($condition:@c2,$sort:@s3,$limit:@l2) { f3, Nested { f4 } }
}

Breaking change: user $pipeline passthrough was removed (raw aggregation escape hatch). A GQL containing $pipeline now fails explicitly instead of being silently ignored. store.aggregate() returns as a strictly opt-in alias (disabled by default, see “Transactions and raw SQL”) — the default behavior is unchanged.

Aggregation

Normalized aggregation lives inside GQL — no separate API, no raw pipeline.

Root-level $group + $having (GROUP BY / HAVING):

rows = await store.query(
    "Course($condition:@c0,$group:@g0,$having:@h0,$sort:@s0,$limit:@l0){ status, n, total }",
    {
        "c0": {"status": {"$ne": "deleted"}},
        "g0": {"by": ["status"], "agg": {"n": {"$count": "*"}, "total": {"$sum": "price"}}},
        "h0": {"n": {"$gt": 1}},
        "s0": {"total": -1},
        "l0": 20,
    },
)

Relation aggregate predicates (semi / anti-join) — filter parents by an aggregate over a relation, without fanning out:

await store.query("Product($condition:@c0,$sort:@s0){ _id, name }", {
    "c0": {
        "$and": [
            {"status": "onSale"},
            {"orders": {"$count": {"$gt": 3}}},                                   # has > 3 orders
            {"$not": {"orders": {"$sum": {"$of": "amount", "$gt": 10000}}}},      # not a whale
        ],
    },
    "s0": {"name": 1},
})

Translates to EXISTS / NOT EXISTS on SQL and $lookup + $match on MongoDB.

Relation-rolling computed columns — declare once in the schema, request by name:

"computes": {
    "itemCount": {"type": "int", "agg": {"$count": "items"}},       # 0 when empty
    "itemsTotal": {"type": "float", "agg": {"$sum": "items.qty"}},  # None when empty
}

Query & write API

items  = await store.query(gql, params)              # list[dict]
one    = await store.query_one(gql, params)          # dict | None
page   = await store.query_with_count(gql, params)   # {'items','total','hasMore','page','pageSize'} (pageSize capped at 5000)
exists = await store.exists("Post", {"_id": pid})
n      = await store.count("Post", {"status": "active"})

doc    = await store.insert("Post", {...})           # auto _id / createdAt / updatedAt
docs   = await store.insert_many("Post", [{...}, ...])
await store.update("Post", {"_id": pid}, {"status": "live"})     # plain fields → $set
await store.update("Post", {"_id": pid}, {"$inc": {"views": 1}}) # '$'-prefixed keys pass through as operators
await store.update_many("Post", {"type": t}, {"status": "live"})
r      = await store.remove("Post", {"_id": pid})    # archives to <collection>_deleted first
await store.mutation("Post", {...})                  # smart upsert + recursive relation children
await store.upsert("Post", {"code": "A1"}, {...})    # explicit-condition upsert (no relation handling)

Notes:

Transactions and raw SQL

async def transfer():
    rows = await store.execute_raw(
        "default", "SELECT * FROM accounts WHERE _id = ? FOR UPDATE", [acc_id])
    await store.execute_raw(
        "default", "UPDATE accounts SET balance = ? WHERE _id = ?", [new_balance, acc_id],
        is_write=True)

await store.transaction("default", transfer)

Concurrency primitives

Declarative task-claiming and compare-and-swap on top of the same plan → Mongo-command pipeline — the patterns that previously required a raw-SQL escape hatch (FOR UPDATE SKIP LOCKED via execute_raw) now have first-class methods, so outbox dispatching and leader election no longer need bare SQL:

got = await store.claim("Job", {"status": "pending", "runAt": {"$lte": now}},
    {"claimedBy": "worker-1", "status": "claimed", "leaseUntil": now + 30_000},
    limit=10, sort=[["runAt", 1]])     # returns the claimed rows (their latest values)

matched = await store.compare_and_set("Leader",
    {"_id": "leader-1", "term": 7},    # current-state guard: the CAS condition
    {"term": 8, "holder": "node-a"})   # the write to apply when it matches

await store.renew_lease("Leader",
    {"_id": "leader-1", "holder": "node-a"},   # ownership condition: only the holder renews
    {"leaseUntil": now + 30_000})

Session (Unit of Work)

async with store.session() as s:
    await s.insert("Order", {...})
    await s.update("Account", cond, {...})
    await s.execute_raw("pg_main", "SELECT ... FOR UPDATE", [1])

DDL generation

sql = store.generate_ddl("mysql")                       # every registered model
sql = store.generate_ddl("postgres", ["Course", "CourseDeleted"])

store.generate_ddl(backend, names=None) maps one registered schema def to one CREATE TABLE — the inverse of sync_schema(), which only reads. The generator is pure text: it never connects to, or writes to, the database (iron rule 6 still holds).

Schema migration (orchestrator)

store.migrate(backend=..., source=..., defs=..., old_defs=..., driver=..., tenant=None, env=None) orchestrates the whitelist migration physically: diff two def versions → per-dialect SQL (generate_migration) → execute in one transaction per table → record a version row in the built-in __schemaMigration table.

r = await store.migrate(
    backend="sqlite", source="default",
    defs={"Note": note_v2},        # {name: defn}, same shape as persist_def
    old_defs=None,                 # None → introspect (needs `driver`: the native driver instance,
                                   #        same convention as sync_schema — the connection view does
                                   #        not expose it); or pass old_defs explicitly to skip it
    driver=db,
)
# r = {"applied": [{"name", "from", "to", "version"}], "skipped": [{"name", "reason"}]}

Multi-datasource connections

A schema is located by (source, database, schema (PostgreSQL only), collection) — the tuple is globally unique across the registry (duplicate registration raises instead of silently mis-routing). The definition file itself carries no location; source / database / schema are resolved from the definitions directory layout plus the connection config.

Location: directory semantics + connection config (definitions carry no location)

A schema definition file contains no location fields (no source / database / schema; namespace is removed). Location is resolved from the definition directory layout plus the connection config:

# Multiple Mongo servers: one source per connection
await init({"mongo_main": db, "pg_a": {"kind": "postgres", "exec": exec}})

# SQL cross-database / PG-schema joins are pushed down natively ("db_a"."t" JOIN "db_b"."t");
# only Mongo cross-db relations fall back to in-memory federation.

Multi-tenant route override — one schema definition, N tenants. Any query/write accepts a { "source", "database", "schema" } override that re-targets commands at execution time (permissions and computed columns still follow the structural schema):

await store.query('User($condition:@c0){...}', params, {"database": "tenant_42"})
await store.insert("Order", data, {"source": "pg_cluster", "schema": "tenant_7"})

route_override is a trusted server-side parameter — it carries no origin check, so forwarding user-controlled input into it lets a caller re-target another tenant’s source/database/schema (CWE-639 authorization-bypass surface). Never pass raw request data here.

Legacy single-db usage (init(db) + schema without location) is unchanged: commands carry source: "default" with the connection’s default database / schema.

Permission context

# Set once per request (in router/dependency layer)
store.set_context({"userId": uid, "roles": ["editor"]})

# Internal/cron jobs — bypass permission checks
await store.run_as_internal(lambda: store.remove("Post", {"_id": pid}))

Fail-secure mode (opt-in)

“No context” can mean both system call and caller forgot the context — by default the latter silently passes every check (fail-open, kept for backward compatibility). For security-sensitive hosts, enable the context requirement once at startup:

store.set_require_context(True)
# now every query/write without a context raises `ERR_NO_CONTEXT:...`
# internal jobs must be explicit:
await store.run_as_internal(lambda: store.remove("Post", {"_id": pid}))

run_as_internal marks the call as {"internal": True}, which is semantically distinct from a missing context and always passes. set_require_context(False) restores the default.

Feedback events

Degraded / pushdown-rejection paths never fail silently — they emit a structured event:

{type, code, layer, message, hint, ...}   # federation_degraded / sql_pushdown_unsupported / ...

store.set_feedback_sink(lambda event: log.warning("store feedback: %s", event))

Schema reference

{
    "name": "Order",
    "collection": "orders",
    "idPrefix": "OD",
    "timestamps": True,                # True (ms, default) | False | "ms" | "s" (seconds); auto-maintain createdAt/updatedAt
    "fields": {
        "_id": "string",                                        # shorthand
        "title": {"type": "string", "default": ""},
        "meta": {"type": "object", "default": {}, "fields": {...}},  # nested object fields
        "bidders": {"type": "array", "default": []},            # array field — inline agg target
    },
    "relations": {
        "items": {"model": "OrderItem", "type": "many",
                  "localField": "_id", "foreignField": "orderId"},
    },
    "computes": {
        "total": {"type": "float", "depends": ["amount"], "fn": lambda d: d["amount"] * 1.1},
        "itemCount": {"type": "int", "agg": {"$count": "items"}},                 # relation agg
        "biddersCount": {"type": "int", "depends": ["bidders"], "agg": {"$size": "bidders"}},
        "maxBid": {"type": "float", "depends": ["bidders"], "agg": {"$max": "bidders.amount"}},
        "bidderCount": {"type": "int", "depends": ["bidders"], "agg": {"$countDistinct": "bidders.bidderId"}},
    },
    "indexes": [
        {"keys": {"status": 1}},
        {"keys": {"code": 1}, "options": {"unique": True}},
    ],
    "read": ["editor", "viewer"],      # optional schema-level role whitelists
    "write": ["editor"],
    "ownerKey": "_id",                 # optional ownership key for the `creator` pseudo-role; default "createdBy"
}

Types: string | int | long | float | double | boolean | array | object | date | any.

Boundary rules worth knowing up front (all fail explicitly, never silently degrade):

Inline array aggregation

An agg whose first path segment is a declared array field (not a relation) aggregates over the document’s own array — compiled into the same pipeline, no extra query phase:

Operator Path Empty array / missing field
$size the array field itself 0
$sum array.element 0
$avg / $max / $min array.element null
$countDistinct array.element 0 — counts distinct non-null values

Rules (violations fail at registration, never silently degrade):

Naming: freeform definitions, system-directed translation

Definitions (collection, fields, referenced relation fields, computed-column keys, fnRef values, index names) may use any style; the engine translates them to the target style. Contract keys (fnRef, localField, foreignField, asyncFn, type, …) and the schema name are never translated.

Target Style Example (orderTotal)
MySQL / PostgreSQL / SQLite (physical) snake_case order_total
MongoDB (physical) camelCase orderTotal
Node.js / Java / C# / Rust (code; computed columns follow) camelCase orderTotal
Go (code; computed columns follow) PascalCase (must be exported) OrderTotal
Python (code; computed columns follow) snake_case order_total

Canonicalization (single implementation core::naming, re-exported by the bindings; hosts must not re-implement it): split on _, -, ., space and at lower/digit-to-upper boundaries; a trailing uppercase in a run followed by a lowercase starts the next token (HTTPServer -> [http, server], userID -> [user, id]); digits stay inside a token (order2Items -> [order2, items]). Reassembly: snake = t1_t2, camel = t1T2, pascal = T1T2.

Two logical names in one schema that canonicalize equal (orderTotal vs order_total) is an error ERR_NAME_CONFLICT: and the service does not start (never silently overwritten). The fields key space is fully open — no field name is reserved: contract keys (type, default, …) live inside a field’s definition object at a different level, so fields: { type: { type: "string" } } is valid. Literal reserved names are only _id (framework primary key) and _score (vector score column).

Computed columns: fnRef binding by composite name + canonical match

Computed columns live at the schema top level, computes: { <key>: { type, fn | asyncFn | agg, fnRef?, depends?, read? } } (fn / asyncFn / agg are mutually exclusive).

Vectors are ordinary field values with a declared dimension and metric: you produce the embedding, the store persists it, ranks by it and returns the score. No embedding model is bundled.

Schema declaration

{
  'name': 'Artwork',
  'collection': 'artworks',
  'fields': {
    'title': {'type': 'string', 'default': ''},
    'caption': {'type': 'string', 'default': ''},
    'embedding': {'type': 'vector', 'dim': 1024, 'metric': 'cosine'},  # 'cosine' | 'l2' | 'ip'
    'embeddingHash': {'type': 'string', 'default': ''},                # optional: reindex digest
  },
  'indexes': [
    # Declares intent for a native approximate index (pgvector / Atlas Search).
    {'keys': {'embedding': 'vector'}, 'options': {'metric': 'cosine', 'method': 'hnsw'}},
  ],
}

Query operator

rows = await store.query(
    'Artwork($condition:@c){ _id, title, _score }',
    {'c': {'embedding': {'$near': {'$vector': [0.12, -0.03], '$limit': 10}}}},
)

Degradation semantics

(await store.capabilities())['vector'] is the static capability report: {supported, native, backend, metrics, maxDim}.

Migration

Transaction boundary

Scenario Atomicity
Single-command API (insert / insert_many / update_many / upsert / remove / count / exists) Naturally atomic within one SQL source (a single statement); single documents are atomic on Mongo
store.transaction(source, fn) Atomic within one SQL source: every command in the scope shares one connection and one transaction; nested same-source scopes use a savepoint (an inner failure rolls back only that scope)
store.session(...) Atomic across multiple calls on one SQL source inside the session; cross-source writes are rejected explicitly (NonAtomicWriteError)
Cross-source multi-write without a session Not atomic (no 2PC / Saga), executed datasource by datasource, and declares nonAtomic via the feedback channel (event non_atomic_write, with the sources)
Multi-step writes on Mongo replica set / sharded: atomic on a single Mongo source (session transaction); standalone: non-atomic and explicitly declares mongo_transaction_unsupported

Triggers

Declarative trigger chains on a schema: write events (insert / update / remove) are expanded by the Rust core at plan time into an ordered list of side-effect steps (plan.triggers) that the host runs in sequence inside the same atomic envelope as the source write. The declaration form (the triggers event keys plus the field table) and the placeholder grammar are defined by the “Triggers” section of the rust-store README; the core’s expanded output is byte-for-byte identical across hosts (golden guard: rust-store/fixtures/triggers/cases.json).

Callback wiring (py_store/crud/triggers.py):

async def _grant_points(args, ctx, host):
    await host['store'].update('User', {'_id': args['userId']}, {'$inc': {'points': args['amount']}})

store.set_trigger_fn('grantPoints', _grant_points)
store.assert_trigger_fns_covered(defns)  # startup: a declared fnRef without an implementation ⇒ ERR_TRIGGER_FN_MISSING

Execution semantics:

Boundaries: single-source real transaction / a cross-source write declares nonAtomic / cross-source writes inside store.session() fail closed (see Transaction boundary); no cascading (a trigger write does not fire further triggers); update_many does not support field-level triggers (explicitly refused, no silent degradation); the command step op: "upsert" is a registration-time Err.

Resource storage (named credential pools)

Content-addressed blob storage beside the CRUD path: resource.put / open / remove / url fan bytes out to a provider registry (local directory, s3; register more via registerProvider / store.register_resource_provider) while metadata lands in three plain CRUD schemas (Resource / ResourceLocation / ResourceBinding).

from py_store import resource

resource.configure({"store": store, "pools": {
    "sys": {"providers": [{"kind": "s3", "options": {...sys creds...}}]},
    "biz": {"providers": [{"kind": "s3", "options": {...biz creds...}}]},
}})
out = await resource.put(bytes=data, file_name="a.pdf", pool="biz")

Transactional capabilities

Capabilities aimed at transactional workloads (orders, inventory — write contention plus complex reads). Full details, semantics and the explicit-error list: doc/transaction-capabilities.md · 中文.

Workflow orchestration (first batch)

Express “orchestration of multi-step data operations” as data: a workflow definition (defn) is pure JSON isomorphic to a schema defn, and each run is persisted to the built-in schema __workflowRun (queryable with plain GQL — zero new observability endpoints). Execution generalizes the existing mutation step-sequence mechanism: linear steps + per-step when guards + fail-fast. The engine lives in the host layer (py_store/workflow.py), core unchanged; aligned with nodejs-store/src/workflow.js (byte-identical outputs guarded by parity anchor tests).

from py_store import workflow

workflow.register({
    'name': 'placeOrder',
    'run': ['admin', 'ops'],              # three-tier whitelists read/write/run (run falls back to write)
    'steps': [
        {'op': 'query', 'as': 'inv',
         'gql': 'Inventory($condition:@c0){_id, stock}',
         'params': {'c0': {'productId': '{{input.productId}}', 'warehouse': '{{input.warehouse}}'}}},
        {'op': 'fail', 'when': {'exists': '{{inv._id}}', 'is': None}, 'message': '库存记录不存在'},
        {'op': 'fail', 'when': {'lt': '{{inv.stock}}', 'than': '{{input.qty}}'}, 'message': '库存不足'},
        {'op': 'mutation', 'model': 'Inventory',
         'data': {'_id': '{{inv._id}}', 'stock': '{{dec:{{inv.stock}},{{input.qty}}}}'}},
    ],
})

run = await workflow.run('placeOrder', {'productId': 'p1', 'warehouse': 'w1', 'qty': 30})
# run['status'] ∈ succeeded | failed | rejected | drySucceeded | dryFailed
# Uniform contract: business failures never raise; the error lives in run['error']
# (set only on failure; always null on success — never `||`-masked downstream)

Explicitly not in the first batch (detected → error; boundaries shipped with the same weight as features)

Not supported Why Escape hatch
Loops / parallel / sub-workflows / human approval DAG & wait semantics explode; linear + when covers the first batch orchestrate in host code via the store API
Auto compensation (Saga) / auto retry Inverse-operation burden; steps have no automatic idempotency inspect run records and handle explicitly
Per-step host callbacks Arbitrary code breaks whitelist governance schema computes (read) / host code (write)
Workflow defn persistence / hot reload Depends on schema versioning (next on the roadmap) defn stays code-side JSON + register, like schemas today
Timers / event triggers Scheduling is a resident-IO concern, orthogonal to pure orchestration call run from the application layer
Placeholders inside gql / array-index paths Injection surface / per-row iteration semantics params binding / host-code orchestration

Scenario examples

The reference sections above describe each capability on its own. The examples below put the same library to work in a concrete business situation — each is copy-runnable against the backends listed in Supported backends.

1. Define a schema as plain Python and query it right away

Scenario. You are adding a Post model to a small service and do not want to hand-write DDL, an ODM class, or a repository layer. The model is data, not code.

from pymongo import AsyncMongoClient
from py_store import init, store

await init(AsyncMongoClient("mongodb://localhost:27017")["mydb"])

store.register({
    "name": "Post",
    "collection": "posts",
    "idPrefix": "PT",
    "fields": {
        "title":  {"type": "string", "default": ""},
        "status": {"type": "string", "default": "draft"},
        "tags":   {"type": "array",  "default": []},
    },
})

await store.insert("Post", {"title": "Hello"})
items = await store.query(
    "Post($condition:@c0,$sort:@s1,$limit:@l) { title, status }",
    {"c0": {"status": "draft"}, "s1": {"createdAt": -1}, "l": 20},
)

Result. items is [{"_id": "PT…", "title": "Hello", "status": "draft"}] — you never wrote status; it is filled by the field default on read.

Why it wins. The model is a plain dict, so there is no class hierarchy, no metaclass and no import-time registration dance — register() plus init() and the model exists. String _ids are generated for you (idPrefix + base36 timestamp + random), and because the very same dict is a valid schema for the Node.js host (nodejs-store), a schema written here is not locked to Python.

2. One schema, swap the backend without touching business code

Scenario. You develop against a local SQLite file and deploy on PostgreSQL. You do not want two query dialects or a rewrite at deploy time.

import aiosqlite
from py_store import executors, init, store

# local dev: a SQLite file
db = await aiosqlite.connect("dev.db")
await init({"default": executors.create_connection("sqlite", db)})

import asyncpg
from py_store import executors, init, store

# production: a PostgreSQL pool
pool = await asyncpg.create_pool(dsn="postgres://user:pass@host:5432/db")
await init({"default": executors.create_connection("postgres", pool)})

The business code that follows is identical in both environments:

items = await store.query(
    "Post($condition:@c0) { title, status }",
    {"c0": {"status": "draft"}},
)

Result. Both environments return the same document shape; only the two datasource lines differ.

Why it wins. GQL is compiled per backend by the shared Rust core — native aggregation on MongoDB, parameterized SQL on SQLite/PostgreSQL/MySQL. Your business code holds no SQL string and no dialect branch, so “change the database” is a change to the connection, not to the query. The local disk datasource (see example 7) slots in the same way with local.connect(...).

3. Fetch a course with its lessons, category and reviews in one query (no N+1)

Scenario. A course detail page needs the course, its category, its lesson list and its reviews. The classic bug is N+1: one query for the course, then one more per relation.

course = await store.query_one(
    "Course($condition:@c0){ _id, title, "
    "  category { _id, name }, "
    "  lessons { _id, title, seq }, "
    "  reviews { _id, score } }",
    {"c0": {"_id": "c1"}},
)

Result. One document with the relations merged in:

{"_id": "c1", "title": "Python入门",
 "category": {"_id": "cat2", "name": "后端"},
 "lessons": [{"_id": "l1", "title": "安装", "seq": 1}, {"_id": "l2", "title": "入门", "seq": 2}],
 "reviews": [{"_id": "r1", "score": 5}, {"_id": "r2", "score": 4}]}

one relations merge into the parent document, many relations become lists ([] when empty, and a missing one is None).

Why it wins. Relations are declared once in the schema, so the selection set alone tells the engine to resolve them: MongoDB compiles to $lookup, SQL to JOIN, in a single query per backend. There is no ORM selectinload / prefetch_related to remember and no $lookup to hand-write, and a missing relation is None/[] rather than a crash. Child lists can even be sorted and paginated per parent: lessons($sort:@s0,$limit:@l0){ _id, title }.

4. Row-level data access for an org / multi-tenant model

Scenario. Online study notes: a student must only read and write their own notes, an audit log is admin-only, and a secret course field is admin-only. You want this enforced in the data layer, not re-implemented in every handler.

store.register({
    "name": "StudyNote", "collection": "study_notes",
    "read": ["creator", "admin"], "write": ["creator", "admin"],
    "fields": {"_id": {"type": "string"}, "title": {"type": "string"},
               "createdBy": {"type": "string"}},
    "relations": {},
})

# set once per request, in the router / dependency layer
store.set_context({"userId": "u1", "roles": ["student"]})

# only u1's notes come back — the owner condition is injected automatically
mine = await store.query("StudyNote($condition:@c0){ _id, title }", {"c0": {"title": "笔记A"}})

Result. mine == [{"_id": "n1", "title": "笔记A"}]. Another user’s note is never returned, and asking for it explicitly returns [] — the owner condition is ANDed with your condition, never replaces it.

Why it wins. creator is a pseudo-role resolved from doc.createdBy == ctx.userId, so ownership is a data fact rather than a per-handler check — schemas that grant it get owner conditions and ownership checks injected on the read and write paths (including count/exists). Field-level whitelists work the same way: a student requesting a field marked read: ["admin"] simply does not get the field, while an explicitly requested, unreadable relation raises store.PermissionError (status = 403) instead of a silent False. Privileged roles (super_admin/admin/internal) pass everything; guest can never write.

5. Computed columns reused at zero cost on the Python side

Scenario. A course list shows revenue = price * enrolledCount and lessonCount = number of lessons. You want both computed at read time, with no denormalized columns to keep in sync.

store.register({
    "name": "Course", "collection": "courses",
    "fields": {"_id": {"type": "string"}, "price": {"type": "float"},
               "enrolledCount": {"type": "int"}},
    "relations": {
        "lessons": {"model": "Lesson", "type": "many",
                    "localField": "_id", "foreignField": "courseId"},
    },
    "computes": {
        # sync compute over a single document
        "revenue": {"type": "float", "depends": ["price", "enrolledCount"],
                    "fn": lambda d: (d.get("price") or 0) * (d.get("enrolledCount") or 0)},
        # relation aggregate: count of the related lessons
        "lessonCount": {"type": "int", "agg": {"$count": "lessons"}},
        # async compute over the whole result page (async def fn(items, ctx) -> None)
        "displayName": {"type": "string", "depends": ["name"], "asyncFn": my_display_name},
    },
})

rows = await store.query(
    "Course($condition:@c0){ _id, price, revenue, lessonCount }",
    {"c0": {"_id": "c2"}},
)

Result. rows == [{"_id": "c2", "price": 100.5, "revenue": 301.5, "lessonCount": 2}]. You never requested enrolledCount, yet revenue is correct — depends pulled the input in — and it does not leak into the output.

Why it wins. Dependency fields are auto-injected for the computation and auto-removed from the projection, so the wire shows exactly what you asked for. A fn is one callback over a row; an asyncFn is one batched pass over the whole result (async def fn(items, ctx)); an agg reuses the same relation-aggregate machinery as query filters, so “count of lessons” needs no hand-written join. Reuse across schemas is name-based: give the column an explicit "fnRef": "common.moneyLabel" and both Python’s common_money_label and Node’s commonMoneyLabel bind to the same logical column (binding is by canonical name; a fnRef with no implementation aborts startup with ERR_FN_MISSING).

6. Aggregation that replaces a hand-rolled SQL report

Scenario. A dashboard needs “courses grouped by status, with a count and a price sum, keeping only groups that have more than one course”. Instead of one SQL query per metric, ask once.

rows = await store.query(
    "Course($condition:@c0,$group:@g0,$having:@h0,$sort:@s0,$limit:@l0){ status, n, total }",
    {
        "c0": {"status": {"$ne": "deleted"}},
        "g0": {"by": ["status"], "agg": {"n": {"$count": "*"}, "total": {"$sum": "price"}}},
        "h0": {"n": {"$gt": 1}},
        "s0": {"total": -1},
        "l0": 20,
    },
)

Result. [{"status": "published", "n": 5, "total": 1000000000000110.8}] — a GROUP BY status ... HAVING count(*) > 1, ordered by the summed price.

Why it wins. $group / $having are part of GQL, so the same call runs native aggregation on MongoDB and a parameterized GROUP BY ... HAVING on SQL — no separate reporting SQL, no dialect branches. The execution order is fixed and documented (WHERE → GROUP BY → HAVING → sort → skip/limit), the operator whitelist is $count / $sum / $avg / $min / $max, and omitting by (or passing []) yields a single whole-table row. Parents can also be filtered by a child aggregate without fan-out — {"orders": {"$count": {"$gt": 3}}} becomes EXISTS / NOT EXISTS on SQL and never duplicates the parent row.

7. Local disk datasource for tests and offline caches

Scenario. You want the same data-layer code in CI or a sandbox with no database to install, and a durable cache that survives a restart — with zero services.

from py_store import init, local, store

await init({"default": local.connect({"dir": "./data/store"})})

store.register({"name": "User", "collection": "users",
               "fields": {"_id": {"type": "string"}, "name": {"type": "string"}},
               "relations": {}})
await store.insert("User", {"name": "Ada"})
rows = await store.query("User{ _id, name }")

Result. rows == [{"_id": "…", "name": "Ada"}], and ./data/store/users.json now holds a JSON array of the documents.

Why it wins. local.connect({"dir": …}) returns an ordinary connection descriptor, so it drops into the same init() and every GQL/CRUD call works unchanged — no SQL engine, no driver (the local source is evaluated by the core’s local evaluator with MongoDB-matching semantics). Writes go through a .tmp file plus an atomic rename, and store.transaction / store.session get snapshot isolation that is atomic by construction. The guardrails are explicit rather than silent: a hard cap of 100,000 documents per collection, no indexes (declaring schema.indexes emits a local_indexes_ignored feedback event), and single-process only.

8. Workflow orchestration: chain multi-step business actions

Scenario. “Place an order” is: read the stock, fail if it is missing, fail if it is insufficient, then decrement. You want the steps and their guards expressed as data, executed fail-fast, and auditable afterwards.

from py_store import workflow

workflow.register({
    "name": "placeOrder",
    "run": ["admin", "ops"],
    "steps": [
        {"op": "query", "as": "inv",
         "gql": "Inventory($condition:@c0){_id, stock}",
         "params": {"c0": {"productId": "{{input.productId}}", "warehouse": "{{input.warehouse}}"}}},
        {"op": "fail", "when": {"exists": "{{inv._id}}", "is": None}, "message": "库存记录不存在"},
        {"op": "fail", "when": {"lt": "{{inv.stock}}", "than": "{{input.qty}}"}, "message": "库存不足"},
        {"op": "mutation", "model": "Inventory",
         "data": {"_id": "{{inv._id}}", "stock": "{{dec:{{inv.stock}},{{input.qty}}}}"}},
    ],
})

run = await workflow.run("placeOrder", {"productId": "p1", "warehouse": "w1", "qty": 30})

Result. run["status"] is one of succeeded | failed | rejected | drySucceeded | dryFailed; business failures do not raise — the reason is in run["error"] (null on success). The run is persisted to the built-in __workflowRun schema, so it is queryable with plain GQL.

Why it wins. The definition is pure JSON isomorphic to a schema defn and is validated statically at register(): an unknown step type, a backward placeholder reference, a when with a missing operand, or a gql with an embedded placeholder all fail at registration (WORKFLOW_UNSUPPORTED) rather than at run time. Steps are a whitelist (query / mutation / fail), placeholders are {{input.x}}, {{step.field}} and {{dec:a,b}}, and every step runs under the caller’s permission context — there is no super-identity. A single-source run is atomic end-to-end; a fail guard records the run as failed with its stepIndex, never silently skipped. workflow.run(name, input, dry_run=True) executes the reads for real and marks writes as wouldRun.

9. Fail-secure mode: one switch closes every open fallback

Scenario. You are going to production and cannot rely on “the caller always remembered to set the context”. You want the safe default to be the enforced default.

from py_store import secure

# once, at startup
secure.secure_mode(admin_roles=["admin"])

Result. With no context set, every query/write now raises ERR_NO_CONTEXT instead of passing (fail-open → fail-secure). Background jobs must opt in explicitly:

await store.run_as_internal(lambda: store.remove("Post", {"_id": pid}))   # still passes

Unconfigured schemas are denied to everyone (admin included) until a whitelist is declared, and registering a new definition without a context is refused. secure.relax_mode() restores the permissive defaults; secure.is_secure() reports the current state.

Why it wins. secure_mode() flips three previously-independent switches together (require-context, unconfigured-roles-closed, meta-policy-closed), so you cannot turn on one and forget the other two. run_as_internal is semantically distinct from a missing context, which is exactly what lets system jobs keep working while forgotten contexts fail closed — the two cases that “no context” used to conflate. Degraded paths stay loud, too: a set_feedback_sink still receives structured events instead of a failure going quiet.

10. Multi-datasource: MySQL + Mongo behind one store, one federated GQL

Scenario. Your users live in MongoDB, your orders live in MySQL, and a profile page needs both. You do not want two clients, two query languages and a hand-written join in the middle.

import asyncmy
from pymongo import AsyncMongoClient
from py_store import executors, init, store

await init({
    "mongo_main": AsyncMongoClient("mongodb://mongo-host:27017")["shop"],
    "mysql_main": executors.create_connection(
        "mysql", await asyncmy.create_pool(host="mysql-host", user="app",
                                           password="…", db="shop", autocommit=True)),
})

# each def pins its home source; the relation is cross-source
store.register({
    "name": "User", "collection": "users", "datasource": "mongo_main",
    "fields": {"_id": {"type": "string"}, "name": {"type": "string"}},
    "relations": {
        "orders": {"model": "Order", "type": "many",
                   "localField": "_id", "foreignField": "userId"}},
})
store.register({
    "name": "Order", "collection": "orders", "datasource": "mysql_main",
    "fields": {"_id": {"type": "string"}, "userId": {"type": "string"},
               "code": {"type": "string"}, "amount": {"type": "float"}},
    "relations": {},
})

rows = await store.query_federated(
    "User($condition:@c0){ _id, name, orders { code, amount } }",
    {"c0": {"name": "ada"}},
)

Result. Each user comes back with its MySQL orders nested in the same document — the store fetches per source, joins in memory, and rehydrates the tree:

[{"_id": "u1", "name": "ada", "orders": [{"code": "A1", "amount": 30.0}, ...]}]

Why it wins. A schema is located by (source, database, collection) and each def carries its datasource, so one registry covers every backend. SQL cross-database / PG-schema joins are pushed down natively ("db_a"."t" JOIN "db_b"."t"); Mongo cross-source relations run through the federated path (query_federated) — and when a segment cannot be pushed down you get a structured federation_degraded event, not a silent slowdown. This exact topology (Mongo users → MySQL orders, plus a PG side) is what the repo’s own four-database stress test runs 8 concurrent workers against (see Performance).

11. Transaction + raw SQL: a transfer that cannot half-land

Scenario. Money moves between two accounts: lock the source row, check the balance, debit one side, credit the other, and append an audit ledger row. Any failure must roll back everything.

async def transfer(amount, from_id, to_id):
    rows = await store.execute_raw(
        "pg_main", "SELECT balance FROM accounts WHERE _id = $1 FOR UPDATE", [from_id])
    if rows["rows"][0]["balance"] < amount:
        raise ValueError("insufficient funds")        # rolls the whole scope back

    await store.execute_raw(
        "pg_main", "UPDATE accounts SET balance = balance - $1 WHERE _id = $2",
        [amount, from_id], is_write=True)
    await store.execute_raw(
        "pg_main", "UPDATE accounts SET balance = balance + $1 WHERE _id = $2",
        [amount, to_id], is_write=True)
    await store.insert("Ledger", {"fromId": from_id, "toId": to_id, "amount": amount})
    #  ↑ a plain CRUD call — it lands on the same transaction connection

await store.transaction("pg_main", lambda: transfer(30, "acc_a", "acc_b"))

Result. All four statements commit together or not at all. execute_raw returns {"rows", "affectedRows"} (rows for reads, the affected count for writes), and store.transaction(source, fn) reuses one connection for everything inside fn — raw SQL and CRUD alike.

Why it wins. Raw SQL is compiled by the core’s raw_stmt_compile (named :name params or positional ?/$1), so escaping and dialect details stay out of your code while the escape hatch stays fully native (FOR UPDATE included). Mongo sources are probed at runtime (replica set / sharded) and wrapped in a session transaction; standalone Mongo or an executor without transaction primitives runs fn as-is and emits mongo_transaction_unsupported / transaction_not_atomic — degradation is announced, never silently pretended. Nested same-source scopes become savepoints: an inner failure rolls back only that scope.

12. Session (Unit of Work): many writes, one commit, fail closed across sources

Scenario. “Create an order” touches three tables: insert the order, decrement stock, append an audit row. You want all of it to behave as one unit of work without hand-managing a connection.

async with store.session() as s:
    order = await s.insert("Order", {"userId": "u1", "amount": 30})
    await s.update("Inventory", {"_id": "sku1"}, {"$inc": {"stock": -1}})
    await s.insert("AuditLog", {"kind": "order", "ref": order["_id"]})
# exits → one commit; any exception → one rollback

Result. Every command on the same SQL source lands on one transaction connection; the session commits once on exit and rolls back as one unit on any exception. A session with no commands never checks out a connection (lazy start).

Why it wins. Sessions nest with savepoints (SAVEPOINT sp_<n>), so an inner failure rolls back only the inner scope while the outer continues. The dangerous case is handled fail-closed: a session writing to ≥2 datasources rolls everything back and raises NonAtomicWriteError — there is no distributed transaction to silently half-commit. Executors without the primitives emit session_not_atomic instead of pretending.

13. DDL generation: schema in, CREATE TABLE text out

Scenario. Your models are registered in code and the DBA (or Alembic) now needs the physical tables. You want a migration starting point without a second source of truth.

sql = store.generate_ddl("postgres")                           # every registered model
sql = store.generate_ddl("sqlite", ["Order", "OrderDeleted"])  # just these two

Result. Plain CREATE TABLE text — never executed, never written to the database — generated from the registered defs:

CREATE TABLE "bench_orders" (
  "_id" TEXT PRIMARY KEY,
  "user_id" TEXT,
  "amount" REAL,
  "qty" INTEGER,
  "status" TEXT,
  "__present" TEXT
);

Why it wins. generate_ddl(backend, names) is the inverse of sync_schema() (which only reads). Freeform field names are translated per backend (userId → user_id on SQL, userId on Mongo), every table gets the __present sentinel column, and the <Model>Deleted archive table is generated exactly once alongside its model. No CREATE INDEX is emitted (SQL backends keep indexes as metadata), and edge cases fail loudly (ddlPresentOverflow) rather than silently.

14. Triggers: a declarative audit trail inside the same transaction

Scenario. Every insert, balance-changing update and delete on Order must leave an audit row — and if the audit write fails, the business write must not land either.

DEFS = [{
    "name": "Order", "collection": "orders", "idPrefix": "OD",
    "fields": {"_id": {"type": "string"}, "amount": {"type": "float"},
               "status": {"type": "string"}},
    "triggers": {
        "insert": [{"name": "aud_ins", "into": "AuditLog", "op": "insert",
                    "data": {"_id": "{{now}}", "kind": "ins",
                             "ref": "{{root._id}}", "amount": "{{root.amount}}"}}],
        "update": [{"name": "aud_upd", "onFields": ["amount"], "into": "AuditLog",
                    "op": "insert",
                    "data": {"_id": "{{now}}", "kind": "upd",
                             "ref": "{{root._id}}", "amount": "{{root.amount}}"}}],
        "remove": [{"name": "aud_del", "fnRef": "auditRemove",
                    "args": {"id": "{{before._id}}"}}],
    },
}, {
    "name": "AuditLog", "collection": "audit_logs", "idPrefix": "AL",
    "fields": {"_id": {"type": "string"}, "kind": {"type": "string"},
               "ref": {"type": "string"}, "amount": {"type": "float"}},
}]

for d in DEFS:
    store.register(d)

async def audit_remove(args, ctx, host):
    await host["store"].insert("AuditLog", {"kind": "del", "ref": args["id"]})

store.set_trigger_fn("auditRemove", audit_remove)
store.assert_trigger_fns_covered(DEFS)   # startup: a declared fnRef without an impl ⇒ ERR_TRIGGER_FN_MISSING

Result. A write event is expanded by the Rust core at plan time into an ordered trigger chain that the host runs inside the same atomic envelope as the source write: insert and update(amount) leave command-written audit rows; removing an order runs the auditRemove callback with the pre-delete document ({{before._id}}).

Why it wins. The chain is data, declared on the schema. update events fire only when the watched fields actually changed (structural deep compare — an unchanged value never fires); {{root.*}} / {{before.*}} / {{now}} are whole-value substitutions (embedding one inside a string raises ERR_TRIGGER_PLACEHOLDER); duplicates (step.name, _id) run once; callbacks receive (args, ctx, {"store": store}) and any store.* inside them hits the same transaction connection. A failing callback or command step rolls the whole write back — and there is no cascading: a trigger write never fires further triggers.

15. Soft delete archive: remove() that keeps the audit trail

Scenario. Compliance wants deletes to keep an audit trail: the row disappears from every list, but the archive stays queryable — including when it was deleted.

r = await store.remove("Order", {"_id": "o1"})
# r == {"deletedCount": 1, "archivedCount": 1}

archived = await store.query("OrderDeleted{ _id, amount, deletedAt }")

Result. The document is copied into the auto-derived OrderDeleted archive (with deletedAt), then deleted from orders. The archive is a normal registered def — plain GQL works on it, and it shows up in generate_ddl() exactly once.

Why it wins. Every registered model gets its <Model>Deleted archive collection/table for free, derived by the Rust core at registration. Archive writes are upsert-by-_id, so re-creating the same _id never collides and a retry after partial failure idempotently re-archives. An empty-condition remove is rejected outright — a full-table delete cannot be triggered by forgetting a filter.

Opting out. A schema that must not keep an archive passes "archive": False at registration — the <Model>Deleted archive def is then never derived (DDL / migrations skip it too), and remove() deletes without archiving (archivedCount == 0, return shape unchanged). For one-shot control pass remove(cond, None, {"archive": False}) to skip the archive for that call only; later default calls archive as usual.

16. FastAPI: per-request permission context in one dependency

Scenario. An internal admin API: paginated list, create, soft-delete. Permissions must be enforced at the data layer for every endpoint, not re-written per route.

from fastapi import Depends, FastAPI, Request
from py_store import init, store

app = FastAPI()

@app.on_event("startup")
async def startup():
    await init({"default": mongo_db})
    store.register({
        "name": "Article", "collection": "articles", "idPrefix": "AR",
        "fields": {"_id": {"type": "string"}, "title": {"type": "string", "default": ""},
                   "body": {"type": "string", "default": ""},
                   "status": {"type": "string", "default": "draft"},
                   "authorId": {"type": "string", "default": ""}},
        "read": ["editor", "viewer", "creator"],
        "write": ["editor", "creator"],
    })

async def current_user(request: Request) -> dict:
    user = {"userId": request.headers["x-user-id"],
            "roles": request.headers.getlist("x-role")}
    store.set_context(user)          # one dependency, whole request covered
    return user

@app.get("/articles")
async def list_articles(page: int = 0, page_size: int = 50,
                        user: dict = Depends(current_user)):
    return await store.query_with_count(
        "Article($sort:@s1){ _id, title, status }",
        {"s1": {"createdAt": -1}, "page": page, "pageSize": page_size})
    # → {"items", "total", "hasMore", "page", "pageSize"}   (pageSize capped at 5000)

@app.delete("/articles/{article_id}")
async def delete_article(article_id: str, user: dict = Depends(current_user)):
    return await store.remove("Article", {"_id": article_id})   # archives first

Result. Role checks (schema/field whitelists, creator ownership) run on every store call with no per-route code; delete leaves an auditable archive row.

Why it wins. The permission context lives in a ContextVar, so setting it in an async dependency covers the whole request path (a sync dependency running in a thread pool would not). A missing context is fail-open by default for backward compatibility; flip store.set_require_context(True) at startup and forgotten contexts raise ERR_NO_CONTEXT instead of passing. Empty-condition writes (update_many / remove with {}) are rejected outright, so a missing filter can never become a full-table write. The full walkthrough lives in doc/use-cases/04-fastapi-admin-backend.md.

Performance

Numbers below come from two sources: a micro-benchmark run on the machine writing this README (SQLite, no services needed — the query/write patterns are copy-runnable) and the repo’s own four-database stress test.

Micro-benchmark (SQLite, measured locally)

Measured 2026-10-15, Python 3.14.4, Windows amd64, file-backed SQLite via aiosqlite, py-store on the Rust core (rust-store-py). Each figure is the median of 4 runs after discarding one warm-up run. Seed data: 200 users; the order table grows to 30,000 rows across the write benchmarks.

Operation Data volume Median Throughput / comparison
insert_many (10 batches × 1,000) 10,000 rows 392 ms ≈ 25,500 rows/s
single insert in a loop 1,000 rows 3,439 ms ≈ 290 rows/s → batching is ≈ 88× faster
GQL tree query — 1 call, relations nested 100 users + 15,000 child rows 517 ms 1 call
hand-written N+1 — same data 100 users + 15,000 child rows 718 ms 101 calls → 1.4× slower
$group pushed down to SQL 10,000 rows 15 ms 1 call
fetch-all + Python-side groupby 10,000 rows 600 ms 41× slower
computed column (read-time fn) 2,000 rows 29 ms ≈ equal to a manual loop (29 ms)

How to read this:

Four-database stress test (recorded in this repo)

stress/stress.py (in this repo) drives MongoDB + MySQL + PostgreSQL + SQLite simultaneously — 8 workers × 200 rounds × 5 ops each (a 3-source federated query, a SQLite write, a PostgreSQL count, a Mongo query, a MySQL $inc update). The recorded run in doc/2026-09-12-压测报告.md reports: 8,000 operations, 0 errors, 824 QPS, federated-query p50 19.3 ms / p95 24.2 ms. (To be verified: not re-run on the machine writing this README — it needs running MySQL/PG/Mongo services; treat these as the repo’s recorded figures, reproducible with the command in that report.)

Where the time goes (design notes)

FAQ

How do I use one schema for both MongoDB and PostgreSQL in Python? Define the schema once as a dict, call init() with your datasource(s), and run the same GQL against either. MongoDB uses native aggregation; MySQL/PostgreSQL/SQLite get parameterized SQL. See Quick start.

How do I query nested / related data without writing $lookup or JOINs? Declare the relation in relations ({"model", "type": "many" | "one", "localField", "foreignField"}) and reference the relation name inside the GQL selection set. It becomes $lookup on Mongo and a JOIN on SQL, returned as nested documents.

Does it support GROUP BY / COUNT / SUM / AVG? Yes — normalized aggregation is part of GQL: root-level $group / $having and relation aggregate predicates. See Aggregation.

Can I filter parents by an aggregate of their children (“products with more than 3 orders”)? Yes — relation aggregate predicates implement semi/anti-join without fanning out; SQL uses EXISTS/NOT EXISTS.

How do I implement row-level permissions? Use store.set_context({"userId": ..., "roles": [...]}) plus schema-level read/write whitelists. The creator pseudo-role adds automatic ownership checks and owner-condition injection (ownership key defaults to createdBy; override per schema with ownerKey, e.g. "ownerKey": "_id" when documents are owned by their own _id). guest can never write. Turn on set_require_context(True) for fail-secure behaviour.

How do I do soft delete? Every registered model automatically gets a <Model>Deleted archive collection/table. store.remove() archives the document first, then deletes it; re-creating the same _id does not collide because the archive write is upsert-by-_id.

Is it usable for multi-tenant applications? Yes. Locate a schema by (source, database, schema, collection) and pass a {"source", "database", "schema"} route override per request. Treat route_override as trusted server-side input only.

Does it run migrations? sync_schema() only reads physical structure via introspection (introspect → merge overlay → register). For applying schema changes there is a whitelist orchestrator: store.migrate() diffs two def versions, renders per-dialect SQL and applies it transactionally per table with a __schemaMigration version trail — whitelist-only (add table/column/index, type widening); anything destructive raises MigrationUnsupported with a two-step recipe instead (see “Schema migration (orchestrator)”). Beyond the whitelist, schema changes remain your migration tool’s job (Alembic, etc.). If you want a starting point, store.generate_ddl(backend) renders CREATE TABLE text from the registered schemas — pure text generation: it never runs or writes DDL by itself.

Can I see the generated query without running it? Yes — store.build_pipeline(gql, params) returns the compiled plan with no execution and no permission/compute application.

What happens when SQL pushdown isn’t possible? The command raises PushdownUnsupportedError and emits a structured feedback event (sql_pushdown_unsupported) through set_feedback_sink. Cross-source pagination/sort degradations emit federation_degraded events. Nothing fails silently.

How is it related to nodejs-store and rust-store? rust-store is the shared Rust engine (GQL parsing, permissions, computed columns, command planning, SQL dialect translation — pure logic, no IO). py-store (pip storepy) and nodejs-store are thin hosts in front of it: they own driver IO, callbacks and placeholder substitution. Same schemas, same GQL, same semantics in Python and Node.

Why is the pip package called storepy and the import py_store? The distribution name on PyPI is storepy; the importable package is py_store. Install with pip install storepy, then from py_store import init, store.

Development

# run the full suite from the repo root (e2e cases auto-skip when MySQL/PG/Mongo are unreachable)
$env:PYTHONPATH='py-store/src'; python -m pytest py-store/tests/ -q

# against custom backends
$env:MYSQL_URI='mysql://user:pass@host:3306/db'; $env:PG_URI='postgres://user:pass@host:5432/db'; $env:MONGO_URI='mongodb://host:27017/db'

Framework usage contract

License

MIT