Python SDK
The official Python SDK for JSONQL. Two lines of configuration give your API dynamic filtering, sorting, pagination, field selection, and relationships — no custom views needed.
| Package | jsonql-py |
| Python | 3.10+ |
| License | MIT |
| Spec | JSONQL v1.1 |
Install
Section titled “Install”pip install jsonql-pyWith framework extras:
pip install "jsonql-py[flask]" # Flask adapterpip install "jsonql-py[fastapi]" # FastAPI adapterpip install "jsonql-py[django]" # Django adapterConfigure & Use
Section titled “Configure & Use”Pick your framework — two lines wire up a complete JSONQL API:
from flask import Flaskfrom jsonql.adapters.flask import create_flask_blueprintfrom jsonql.adapters.options import AdapterOptions
app = Flask(__name__)
opts = AdapterOptions(execute=lambda sql, params: db.execute(sql, params).fetchall())bp = create_flask_blueprint(opts)app.register_blueprint(bp, url_prefix="/api")
app.run(port=8080)FastAPI
Section titled “FastAPI”from fastapi import FastAPIfrom jsonql.adapters.fastapi import create_fastapi_routerfrom jsonql.adapters.options import AdapterOptions
app = FastAPI()
opts = AdapterOptions(execute=lambda sql, params: db.execute(sql, params).fetchall())router = create_fastapi_router(opts)app.include_router(router, prefix="/api")Django
Section titled “Django”# urls.pyfrom jsonql.adapters.django import create_django_urlsfrom jsonql.adapters.options import AdapterOptions
opts = AdapterOptions(execute=lambda sql, params: cursor.execute(sql, params).fetchall())
urlpatterns = [ path("api/", include(create_django_urls(opts))),]That’s it. One execute function, one router mount. Your clients can now query any table dynamically.
What Your Clients Can Do
Section titled “What Your Clients Can Do”Every query is a JSON POST to /api/{table}:
# Select specific fields, filter, sort, paginatecurl -X POST http://localhost:8080/api/users -H 'Content-Type: application/json' -d '{ "fields": ["id", "name", "email"], "where": { "status": { "eq": "active" } }, "sort": ["-created_at"], "limit": 20}'# → { "data": [{ "id": 1, "name": "Alice", "email": "alice@co.com" }, ...] }# Complex filterscurl -X POST http://localhost:8080/api/products -d '{ "where": { "and": [ { "price": { "gte": 10, "lte": 100 } }, { "category": { "in": ["electronics", "books"] } } ] }, "sort": ["price"], "limit": 50}'# Include related datacurl -X POST http://localhost:8080/api/users -d '{ "fields": ["id", "name"], "include": { "posts": { "fields": ["title", "created_at"], "limit": 5 } }}'# Aggregation & groupBycurl -X POST http://localhost:8080/api/orders -d '{ "aggregate": { "total": { "fn": "sum", "field": "amount" } }, "groupBy": ["status"]}'CRUD mutations:
# Createcurl -X POST http://localhost:8080/api/users \ -d '{ "data": { "name": "Bob", "email": "bob@co.com" } }'
# Updatecurl -X PATCH http://localhost:8080/api/users \ -d '{ "patch": { "status": "inactive" }, "where": { "id": { "eq": 1 } } }'
# Deletecurl -X DELETE http://localhost:8080/api/users \ -d '{ "where": { "id": { "eq": 1 } } }'Adding Schema (Optional)
Section titled “Adding Schema (Optional)”Without schema, all columns are queryable. With schema, you control field exposure and enable relationship resolution:
from jsonql.schema import Schema, Table, Field, Relation
schema = Schema(tables={ "users": Table( fields={ "id": Field(type="number"), "name": Field(type="string"), "email": Field(type="string"), }, relations={ "posts": Relation(type="hasMany", table="posts", field="user_id"), }, ),})
opts = AdapterOptions(execute=run_sql, schema=schema)Lifecycle Hooks
Section titled “Lifecycle Hooks”Inject tenant isolation, audit logging, or authorization:
def before_query(query: dict, table: str) -> dict: """Add tenant filter to every query.""" query["where"] = { "and": [query.get("where", {}), {"tenant_id": {"eq": current_tenant()}}] } return query
opts = AdapterOptions( execute=run_sql, before_query=before_query, after_query=lambda result, table: audit_log(table, len(result)),)Supported Databases
Section titled “Supported Databases”| Database | Dialect | Python Driver |
|---|---|---|
| PostgreSQL | postgres | psycopg2 / asyncpg |
| MySQL | mysql | mysql-connector-python / pymysql |
| SQLite | sqlite | sqlite3 (stdlib) |
| MSSQL | mssql | pyodbc / pymssql |
| MongoDB | mongodb | pymongo |
Dialect is auto-detected from connection; explicit setting is optional:
opts = AdapterOptions(execute=run_sql, dialect="postgres")Error Handling
Section titled “Error Handling”All errors inherit from JsonQLError with an error_code attribute:
graph TD E["JsonQLError"] --> P["JsonQLParseError<br/>(PARSE_ERROR)"] E --> V["JsonQLValidationError<br/>(VALIDATION_ERROR)"] E --> T["JsonQLTranspileError<br/>(TRANSPILE_ERROR)"] E --> X["JsonQLExecutionError<br/>(EXECUTION_ERROR)"]from jsonql.errors import JsonQLError, JsonQLValidationError
try: result = engine.execute("users", query)except JsonQLValidationError as e: print(e.error_code) # "VALIDATION_ERROR"except JsonQLError as e: print(e.error_code) # Any JSONQL errorAdapter error responses:
{ "error": "Field 'secret' is not allowed", "error_code": "VALIDATION_ERROR" }Advanced: Engine (Direct Use)
Section titled “Advanced: Engine (Direct Use)”For custom pipelines outside the framework adapters:
from jsonql import JsonQLEngine
engine = JsonQLEngine(execute=run_sql, dialect="postgres", schema=schema)
result = engine.execute("users", { "where": {"status": {"eq": "active"}}, "fields": ["id", "name"], "limit": 10,})# result["data"] → list of dicts# result["is_mutation"] → boolAdvanced: Query Builder
Section titled “Advanced: Query Builder”For server-side programmatic query construction:
from jsonql.builder import QueryBuilder
query = ( QueryBuilder("users") .select("id", "name", "email") .where({"status": {"eq": "active"}}) .order_by("-created_at") .limit(10) .build())With condition helpers:
from jsonql.builder import eq, gt, in_list
query = ( QueryBuilder("products") .where({"price": gt(10), "category": in_list(["books", "electronics"])}) .build())Core API
Section titled “Core API”| Class | Purpose |
|---|---|
AdapterOptions | Configure adapter: execute, schema, hooks, dialect |
create_flask_blueprint() | Flask route blueprint |
create_fastapi_router() | FastAPI route factory |
JsonQLEngine | Manual transpile-and-execute pipeline |
Parser | Parse & validate incoming JSON |
Transpiler | Convert parsed query → SQL + params |
Hydrator | Convert flat rows → nested dicts |
QueryBuilder | Fluent query construction (advanced) |
Compliance
Section titled “Compliance”135/135 tests passing across all configurations:
| Adapter | PostgreSQL | MySQL | SQLite | MSSQL | MongoDB |
|---|---|---|---|---|---|
| Flask | ✅ | ✅ | ✅ | ✅ | ✅ |
| FastAPI | ✅ | ✅ | ✅ | ✅ | ✅ |
| Django | ✅ | ✅ | ✅ | ✅ | ✅ |
Development
Section titled “Development”pytest # Run all testsruff check . # Lint (rules E/F/I/W)mypy src/ # Type checking (strict)