Skip to content
BranchesTags

Repository files navigation

querymesh

A production-grade, universal database abstraction library for Python.

querymesh gives you a single consistent API across MySQL, PostgreSQL, SQLite, and SQL Server with transparent query caching, a built-in analytics engine, horizontal sharding, read/write splitting, async support, vector search, visualizers, a self-hosted operational store, and a full suite of production resilience tools - all in one library with zero mandatory dependencies beyond the standard library.

Contents

Features

Core

  • Unified DatabaseClient API across all supported engines
  • Transparent query caching (in-memory LRU or Redis) with TTL and tag-based invalidation
  • SQL normalization for consistent cache key generation
  • Query analytics: execution time, slow query detection, cache hit/miss ratio

Developer Experience

  • Fluent query builder (Select, Insert, Update, Delete)
  • Type-safe model coercion (dataclasses and Pydantic v1/v2)
  • fetch_page() for built-in pagination
  • Schema migration runner with rollback support
  • MockAdapter and RecordingAdapter for zero-setup testing
  • Event hooks for every query lifecycle stage

Async

  • AsyncDatabaseClient with the same API as the sync client
  • Async adapters for SQLite (aiosqlite), PostgreSQL (asyncpg), MySQL (aiomysql)

Resilience

  • RetryPolicy with exponential backoff and jitter
  • CircuitBreaker (CLOSED / OPEN / HALF-OPEN)
  • ConnectionPool with checkout timeout and idle eviction
  • Cache stampede protection via SingleFlight
  • Cache warming (CacheWarmer)

Scaling

  • ScalingEngine - drop-in replacement for DatabaseClient that distributes queries
  • Read/write splitting across primary + N read replicas
  • Horizontal sharding (HASH / MODULO / RANGE strategies)
  • Scatter/gather for cross-shard queries (parallel, ThreadPoolExecutor)
  • 5 load balancing strategies: round-robin, least-connections, weighted-random, latency-aware, random
  • Per-instance circuit breakers and automatic retry on instance failure
  • Background HealthMonitor with configurable thresholds and state-change callbacks
  • Runtime add_replica() / remove_replica() / scale_out() without restart

Vector Operations

  • High-level VectorStore and AsyncVectorStore APIs
  • Pluggable embedding models: RandomEmbedding, OpenAIEmbedding, SentenceTransformerEmbedding, BedrockEmbedding, AzureOpenAIEmbedding, CachedEmbeddingModel
  • Three backends: NumpyBackend (in-memory), SqliteBackend (BLOB), PgVectorBackend (pgvector)
  • Distance metrics: cosine, euclidean, dot product, manhattan - with NumPy acceleration
  • Metadata filtering and delete_where for vector lifecycle management

Cloud Integrations

  • AWS: RDSIAMAdapter (auto-refreshing IAM token auth), GlueAdapter (Amazon Athena/Glue)
  • Azure: AzureADAdapter (Azure AD token auth for SQL, PostgreSQL, MySQL)
  • Google Cloud: BigQueryAdapter

Visualizers

  • QueryTracer - captures per-query lifecycle as a FlowGraph from the EventBus
  • AsciiRenderer - terminal-friendly flow diagram, zero dependencies
  • MermaidRenderer - Mermaid flowchart TD syntax, paste anywhere
  • HtmlFlowRenderer - colour-coded HTML fragment with inline CSS
  • DashboardExporter - self-contained HTML analytics dashboard (Chart.js)
  • LiveServer - optional live HTTP dashboard with auto-refresh (pip install flask)

Internal Store

  • Self-hosted persistence in a _qm_* schema on the connected DB
  • Logs: query execution log, slow query log, error log, optional audit trail
  • Configurable: choose what to store, set row limits and auto-pruning
  • PolicyStore - runtime-editable key-value config persisted in the DB
  • Ships with built-in default policies (cache_ttl, slow_query_threshold_ms, ...)
  • Auto-wires to EventBus for zero-config logging

Observability

  • Per-instance metrics: total queries, error rate, avg/p95 latency, active connections
  • JSONFileExporter for persistent analytics snapshots
  • Optional PrometheusExporter (pip install prometheus-client)
  • engine.stats() full topology snapshot

Architecture

  • Adapter pattern - add new engines without touching core logic
  • Strategy pattern for swappable cache backends
  • Dependency injection for every component
  • Full type-hint coverage

Supported Databases

EngineSync AdapterAsync AdapterExtra Install
SQLiteSqliteAdapterAsyncSqliteAdapterpip install aiosqlite (async only)
MySQLMySQLAdapterAsyncMySQLAdapterpip install querymesh[mysql]
PostgreSQLPostgreSQLAdapterAsyncPostgreSQLAdapterpip install querymesh[postgresql]
SQL ServerSQLServerAdapter-pip install querymesh[sqlserver]
BigQueryBigQueryAdapter-pip install google-cloud-bigquery
AWS AthenaGlueAdapter-pip install pyathena

Installation

pip install querymesh
# Database-specific drivers
pip install querymesh[mysql]
pip install querymesh[postgresql]
pip install querymesh[sqlserver]
# Async drivers
pip install aiosqlite # async SQLite
pip install asyncpg # async PostgreSQL
pip install aiomysql # async MySQL# Optional extras
pip install querymesh[redis] # Redis cache backend
pip install querymesh[viz] # Live dashboard server (Flask)
pip install prometheus-client # Prometheus metrics export
pip install numpy # accelerated vector distance functions
pip install sentence-transformers # local embedding model# Cloud SDKs
pip install boto3 # AWS (RDS IAM, Bedrock)
pip install azure-identity # Azure AD auth
pip install google-cloud-bigquery # BigQuery
pip install openai # OpenAI embeddings# Everything
pip install querymesh[all]

Quick Start

fromquerymeshimportDatabaseClientfromquerymesh.adapters.sqliteimportSqliteAdapterfromquerymesh.configimportDatabaseConfig, CacheConfigclient=DatabaseClient(
adapter=SqliteAdapter(DatabaseConfig(database=":memory:")),
cache_config=CacheConfig(ttl=300),
)
withclient:
client.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
client.execute("INSERT INTO users VALUES (?, ?, ?)", params=(1, "Alice", 30))
row=client.fetch_one("SELECT * FROM users WHERE id = ?", params=(1,))
rows=client.fetch_all("SELECT * FROM users WHERE age > ?", params=(18,))
print(row) # {'id': 1, 'name': 'Alice', 'age': 30}

Core Client

fromquerymeshimportDatabaseClientfromquerymesh.adapters.postgresqlimportPostgreSQLAdapterfromquerymesh.configimportDatabaseConfig, CacheConfig, AnalyticsConfigclient=DatabaseClient(
adapter=PostgreSQLAdapter(DatabaseConfig(
host="localhost", port=5432,
user="app", password="secret", database="mydb",
)),
cache_config=CacheConfig(ttl=60, backend="redis", redis_url="redis://localhost:6379"),
analytics_config=AnalyticsConfig(slow_query_threshold_ms=200),
)
withclient:
# Write - never cachedclient.execute("INSERT INTO orders (user_id, total) VALUES (%s, %s)", params=(1, 99.99))
# Read - result cached for 60 sorders=client.fetch_all("SELECT * FROM orders WHERE user_id = %s", params=(1,))
# Bypass cache read, but still refresh the cache with the fresh resultorders=client.fetch_all("SELECT * FROM orders", bypass_cache=True)
# Per-call TTL overriderow=client.fetch_one("SELECT * FROM users WHERE id = %s", params=(1,), ttl=3600)
# Tag-based invalidation - invalidate all queries tagged "orders" in one callorders=client.fetch_all("SELECT * FROM orders", tags=["orders"])
client.invalidate_tag("orders")
# Explicit key invalidationclient.invalidate_query("SELECT * FROM orders WHERE user_id = %s", params=(1,))

Async Client

importasynciofromquerymeshimportAsyncDatabaseClientfromquerymesh.adapters.async_sqliteimportAsyncSqliteAdapterfromquerymesh.configimportDatabaseConfig, CacheConfigasyncdefmain():
client=AsyncDatabaseClient(
adapter=AsyncSqliteAdapter(DatabaseConfig(database=":memory:")),
cache_config=CacheConfig(ttl=60),
)
asyncwithclient:
awaitclient.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, val TEXT)")
awaitclient.execute("INSERT INTO t VALUES (?, ?)", params=(1, "hello"))
row=awaitclient.fetch_one("SELECT * FROM t WHERE id = ?", params=(1,))
rows=awaitclient.fetch_all("SELECT * FROM t")
asyncwithclient.transaction():
awaitclient.execute("UPDATE t SET val = ? WHERE id = ?", params=("world", 1))
asyncio.run(main())

Caching

querymesh caches SELECT queries transparently. The cache key is derived from the normalized SQL + parameters + database context, so identical queries with different params produce different keys.

fromquerymesh.configimportCacheConfig# In-memory LRU (default)CacheConfig(backend="memory", ttl=300, max_size=1000)
# RedisCacheConfig(backend="redis", ttl=300, redis_url="redis://localhost:6379/0")
# Disable caching entirelyCacheConfig(enabled=False)
# Cache all queries, not just SELECTCacheConfig(cache_select_only=False)

Tag-based invalidation lets you logically group cache entries and flush them together:

# Store with tagsclient.fetch_all("SELECT * FROM users", tags=["users"])
client.fetch_all("SELECT * FROM users WHERE active = ?", params=(True,), tags=["users"])
# Flush everything tagged "users" in one callclient.invalidate_tag("users")

Transactions

withclient.transaction():
client.execute("UPDATE accounts SET balance = balance - 100 WHERE id = ?", (1,))
client.execute("UPDATE accounts SET balance = balance + 100 WHERE id = ?", (2,))
# Committed on clean exit, rolled back on any exception
# Asyncasyncwithclient.transaction():
awaitclient.execute("UPDATE accounts SET balance = balance - 100 WHERE id = ?", (1,))
awaitclient.execute("UPDATE accounts SET balance = balance + 100 WHERE id = ?", (2,))

Batch Execution & Bulk Insert

execute_many — driver-native executemany for arbitrary statements:

rows= [("Alice", 30), ("Bob", 25), ("Carol", 35)]
client.execute_many(
"INSERT INTO users (name, age) VALUES (?, ?)",
param_list=rows,
)

bulk_insert — optimized multi-row VALUES batching for SQLite and MySQL (falls back to execute_many for PostgreSQL / SQL Server):

inserted=client.bulk_insert(
table="events",
columns=["type", "ts", "payload"],
rows=[
("click", 1700000000, "{}"),
("view", 1700000001, "{}"),
# ... thousands more
],
batch_size=500, # rows per SQL statementon_conflict="OR IGNORE", # optional conflict clause (SQLite)
)
print(inserted) # total rows inserted

Async version has the same signature:

inserted=awaitclient.bulk_insert("events", columns, rows, batch_size=500)

Query Streaming

Stream large result sets row-by-row without loading them all into memory:

forrowinclient.stream("SELECT * FROM logs WHERE ts > ?", params=(cutoff,), chunk_size=200):
process(row)
  • SQLite: cursor.fetchmany() loop — constant memory regardless of result size
  • MySQL: unbuffered server-side cursor
  • PostgreSQL: named server-side cursor
  • Other adapters: falls back to fetch_all row-by-row yield

With model coercion:

@dataclassclassLogEntry:
id: intmessage: strforentryinclient.stream("SELECT * FROM logs", model=LogEntry):
print(entry.message)

Async:

asyncforrowinclient.stream("SELECT * FROM logs"):
awaitprocess(row)

Streaming bypasses the cache. Analytics and events are recorded based on total wall-clock time of the stream.

Paginated Queries

fromquerymesh.paginationimportPageResultresult=client.fetch_page(
"SELECT * FROM products ORDER BY created_at DESC",
page=2,
page_size=25,
)
print(result.rows) # list of dicts for this pageprint(result.total) # total rows across all pagesprint(result.total_pages) # ceil(total / page_size)print(result.has_next) # True if more pages existprint(result.to_dict()) # serialisable summary

Query Builder

A fluent, database-agnostic builder that produces parameterized (sql, params) tuples.

fromquerymesh.builderimportSelect, Insert, Update, Delete# SELECTsql, params= (
Select("users")
.columns("id", "name", "email")
.where("active = ?", True)
.where("age > ?", 18)
.order_by("name ASC")
.limit(10)
.offset(20)
.build()
)
rows=client.fetch_all(sql, params=params)
# INSERTsql, params=Insert("users").values(name="Alice", age=30).build()
client.execute(sql, params=params)
# UPDATEsql, params=Update("users").set(active=False).where("last_login < ?", cutoff).build()
client.execute(sql, params=params)
# DELETEsql, params=Delete("sessions").where("expired = ?", True).build()
client.execute(sql, params=params)
# MySQL / PostgreSQL - use %s placeholdersql, params=Select("users").where("id = ?", 1).build(placeholder="%s")

Builder supports: distinct(), join(), group_by(), having(), order_by(), limit(), offset().

Type-Safe Models

Pass any dataclass or Pydantic model to fetch_one / fetch_all and rows are coerced automatically.

fromdataclassesimportdataclass@dataclassclassUser:
id: intname: strage: intusers=client.fetch_all("SELECT * FROM users", model=User)
# → list[User]user=client.fetch_one("SELECT * FROM users WHERE id = ?", params=(1,), model=User)
# → User | None

Works with Pydantic v2 (model_validate), Pydantic v1 (parse_obj), and any dataclass.

Event Hooks

Subscribe to query lifecycle events without modifying client code.

fromquerymesh.hooksimportEventBus, QueryExecutedEvent, SlowQueryEvent, QueryErrorEventbus=EventBus()
@bus.on(SlowQueryEvent)defalert_slow(event: SlowQueryEvent):
print(f"SLOW {event.execution_time_ms:.0f}ms: {event.query[:80]}")
@bus.on(QueryErrorEvent)deflog_error(event: QueryErrorEvent):
print(f"ERROR: {event.error} - {event.query[:80]}")
client=DatabaseClient(
adapter=...,
event_bus=bus,
slow_query_threshold_ms=200,
)

Available events:QueryExecutedEvent, CacheHitEvent, CacheMissEvent, SlowQueryEvent, QueryErrorEvent, VectorSearchEvent, EmbeddingGeneratedEvent.

Analytics & Export

stats=client.analytics.get_stats()
# {# "total_queries": 1042,# "unique_queries": 18,# "cache_hits": 830,# "cache_misses": 212,# "cache_hit_ratio": 0.797,# "avg_execution_time_ms": 3.2,# }slow=client.analytics.get_slow_queries(threshold_ms=100)
top=client.analytics.get_top_queries(limit=10)

JSON persistence - append snapshots to a JSONL file:

fromquerymesh.analytics.exportersimportJSONFileExporterexporter=JSONFileExporter(
client.analytics,
path="querymesh_analytics.jsonl",
auto_flush_interval_s=60, # optional background flush
)
exporter.flush() # write immediatelyexporter.stop() # cancel background timer

Prometheus metrics (requires pip install prometheus-client):

fromquerymesh.analytics.exportersimportPrometheusExporterprom=PrometheusExporter(client.analytics, namespace="myapp")
prom.update() # push current stats to gauges

Retry & Circuit Breaker

fromquerymesh.retryimportRetryPolicy, CircuitBreakerfromquerymesh.configimportRetryConfig, CircuitBreakerConfig# Retry with exponential backoff + jitterpolicy=RetryPolicy(RetryConfig(
max_attempts=3,
base_backoff_ms=100,
backoff_multiplier=2.0,
max_backoff_ms=5000,
jitter=True,
))
result=policy.execute(adapter.fetch_all, "SELECT * FROM users")
# Circuit breaker - stops hammering a failing servicebreaker=CircuitBreaker(CircuitBreakerConfig(
failure_threshold=5, # open after 5 consecutive failuresrecovery_timeout_s=60.0, # probe again after 60 ssuccess_threshold=2, # 2 successes to close
))
result=breaker.call(adapter.fetch_all, "SELECT * FROM users")

Connection Pool

fromquerymesh.poolimportConnectionPoolfromquerymesh.adapters.sqliteimportSqliteAdapterfromquerymesh.configimportDatabaseConfig, PoolConfigpool=ConnectionPool(
factory=lambda: SqliteAdapter(DatabaseConfig(database="app.db")),
config=PoolConfig(
min_size=2,
max_size=10,
checkout_timeout=30.0,
max_idle_seconds=300,
),
)
withpool.acquire() asadapter:
rows=adapter.fetch_all("SELECT * FROM users")
pool.close()

Schema Migrations

migrations/
001_create_users.sql
001_create_users.down.sql <- optional rollback
002_add_email_column.sql
002_add_email_column.down.sql

fromquerymesh.migrationsimportMigrationRunnerfromquerymesh.adapters.sqliteimportSqliteAdapterfromquerymesh.configimportDatabaseConfigadapter=SqliteAdapter(DatabaseConfig(database="app.db"))
adapter.connect()
runner=MigrationRunner(adapter, migrations_dir="./migrations")
result=runner.migrate() # apply all pending migrationsprint(result.applied) # ["001_create_users", "002_add_email_column"]runner.rollback() # undo the last applied migrationprint(runner.applied()) # ["001_create_users"]print(runner.pending()) # ["002_add_email_column"]

Migrations are tracked in a _qm_migrations table created automatically in your database.

Schema Introspection

Read live schema metadata (tables, columns, indexes) from a connected database:

schema=client.inspect()
print(schema.engine) # "sqlite"print(schema.table_names()) # ["orders", "products", "users"]fortableinschema.tables:
print(f"\n{table.name} ({table.row_count} rows)")
forcolintable.columns:
pk=" PK"ifcol.primary_keyelse""null=""ifcol.nullableelse" NOT NULL"print(f" {col.name:<25}{col.data_type}{pk}{null}")
foridxintable.indexes:
print(f" [index] {idx.name} ({', '.join(idx.columns)})")

Include row counts per table (runs one COUNT(*) per table):

schema=client.inspect(include_row_counts=True)

Use SchemaInspector directly against any adapter:

fromquerymesh.introspectionimportSchemaInspectorinspector=SchemaInspector(adapter)
schema=inspector.inspect()
table=schema.table("users") # → TableInfo | None

Supported engines: SQLite, PostgreSQL, MySQL. Other adapters raise NotImplementedError.

Cache Warming & Stampede Protection

Warm the cache before traffic hits to prevent cold-start latency spikes:

fromquerymesh.cache.warmerimportCacheWarmer, WarmupQuerywarmer=CacheWarmer(client)
result=warmer.warm([
WarmupQuery("SELECT * FROM config", ttl=3600),
WarmupQuery("SELECT * FROM categories", ttl=1800),
WarmupQuery("SELECT * FROM products LIMIT 100", tags=["products"]),
])
print(f"Warmed {result.succeeded}/{result.total} in {result.duration_ms:.0f}ms")

Stampede protection is built into DatabaseClient via SingleFlight. When multiple threads simultaneously miss the same cache key, only one fetches from the database. All others wait and read the freshly-written value - no thundering herd.

Scaling Engine

ScalingEngine is a drop-in replacement for DatabaseClient that distributes queries across multiple database instances.

Read/Write Splitting

fromquerymeshimportScalingEngine, ScalingConfigfromquerymesh.scalingimportLoadBalancingStrategyfromquerymesh.adapters.postgresqlimportPostgreSQLAdapterfromquerymesh.configimportDatabaseConfigprimary=PostgreSQLAdapter(DatabaseConfig(host="db-primary", ...))
replica1=PostgreSQLAdapter(DatabaseConfig(host="db-replica-1", ...))
replica2=PostgreSQLAdapter(DatabaseConfig(host="db-replica-2", ...))
engine=ScalingEngine.from_replicas(
primary=primary,
replicas=[replica1, replica2],
config=ScalingConfig(
load_balancing=LoadBalancingStrategy.LEAST_CONNECTIONS,
health_check_interval_s=10.0,
retry_on_failure=True,
),
)
withengine:
# SELECT -> load-balanced across replicasusers=engine.fetch_all("SELECT * FROM users WHERE active = %s", params=(True,))
# INSERT / UPDATE / DELETE / DDL -> always the primaryengine.execute("INSERT INTO events (user_id, action) VALUES (%s, %s)", params=(1, "login"))
# Force a read onto the primary (read-your-writes)user=engine.fetch_one("SELECT * FROM users WHERE id = %s", params=(1,), force_primary=True)
withengine.transaction():
engine.execute("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
engine.execute("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))

Horizontal Sharding

fromquerymeshimportScalingEnginefromquerymesh.scalingimportShardingStrategy, RangeRule, ShardMap# Hash sharding - uniform distributionengine=ScalingEngine.from_shards(
{0: adapter_us, 1: adapter_eu, 2: adapter_ap, 3: adapter_sa},
strategy=ShardingStrategy.HASH,
)
withengine:
# Routed to correct shard via shard_keyorders=engine.fetch_all(
"SELECT * FROM orders WHERE user_id = %s",
params=(user_id,),
shard_key=user_id,
)
# No shard_key -> scatter/gather: runs on ALL shards in paralleltotal=engine.fetch_one("SELECT COUNT(*) AS cnt FROM orders")

Health Monitoring

fromquerymesh.scalingimportHealthMonitor, InstanceStatedefon_change(instance, old_state, new_state):
ifnew_state==InstanceState.DOWN:
alert_on_call(instance.instance_id)
engine=ScalingEngine.from_replicas(
primary=primary_adapter,
replicas=[r1, r2],
config=ScalingConfig(
health_check_interval_s=15.0,
degraded_threshold=1,
down_threshold=3,
recovery_threshold=2,
),
on_instance_state_change=on_change,
)

Dynamic Scaling

new_replica=PostgreSQLAdapter(DatabaseConfig(host="db-replica-4", ...))
new_replica.connect()
engine.add_replica(new_replica, instance_id="replica-4", weight=2.0)
engine.remove_replica("replica-1")
engine.scale_out([adapter_r4, adapter_r5, adapter_r6])

Vector Operations

High-level vector store with pluggable embedding models and storage backends.

fromquerymesh.vectorimportVectorStore, DistanceMetric, RandomEmbeddingstore=VectorStore(
embedding_model=RandomEmbedding(dimensions=384),
dimensions=384,
metric=DistanceMetric.COSINE,
)
# Insert by text (embedding generated automatically)store.upsert_text("doc-1", "the quick brown fox", metadata={"category": "animals"})
store.upsert_text("doc-2", "python programming tips", metadata={"category": "tech"})
store.upsert_text("doc-3", "fast running cheetah", metadata={"category": "animals"})
# Similarity searchresults=store.search_text("speedy wildlife", top_k=2)
forrinresults:
print(r.rank, r.score, r.id, r.metadata)
# Metadata filterresults=store.search_text("animals", filter={"category": "animals"})
# Direct vector insertstore.upsert("vec-1", [0.1, 0.2, ...], metadata={"source": "manual"})
# Lifecyclerec=store.get("doc-1")
store.delete("doc-1")
n=store.delete_where({"category": "tech"})
print(store.count())

Backends:

fromquerymesh.vector.backends.sqlite_backendimportSqliteBackend# In-memory NumPy (default, zero deps)store=VectorStore(dimensions=384)
# SQLite BLOB persistencebackend=SqliteBackend(adapter=sqlite_adapter, dimensions=384)
backend.create_table()
store=VectorStore(backend=backend, dimensions=384)
# PostgreSQL pgvector (pip install pgvector)fromquerymesh.vector.backends.pgvector_backendimportPgVectorBackendbackend=PgVectorBackend(adapter=pg_adapter, dimensions=1536)
store=VectorStore(backend=backend, dimensions=1536)

Embedding models:

fromquerymesh.vectorimport (
RandomEmbedding, # testing/prototyping - no depsOpenAIEmbedding, # pip install openaiSentenceTransformerEmbedding, # pip install sentence-transformersBedrockEmbedding, # pip install boto3AzureOpenAIEmbedding, # pip install openaiCachedEmbeddingModel, # wraps any model with an LRU cache
)
model=CachedEmbeddingModel(OpenAIEmbedding(api_key="sk-..."), max_size=1000)
store=VectorStore(embedding_model=model, dimensions=1536)

Async:

fromquerymesh.vectorimportAsyncVectorStorestore=AsyncVectorStore(embedding_model=model, dimensions=384)
awaitstore.upsert_text("a", "hello async world")
results=awaitstore.search_text("hello", top_k=3)

Cloud Integrations

AWS RDS IAM Authentication

fromquerymesh.cloud.awsimportRDSIAMAdapterfromquerymesh.configimportDatabaseConfigadapter=RDSIAMAdapter(
base_config=DatabaseConfig(
host="mydb.cluster-xxx.us-east-1.rds.amazonaws.com",
port=5432,
user="app_user",
database="mydb",
),
region="us-east-1",
engine="postgresql", # or "mysql"
)
adapter.connect()
rows=adapter.fetch_all("SELECT * FROM users LIMIT 10")

Tokens are generated via boto3 and refreshed automatically before expiry.

AWS Athena / Glue

fromquerymesh.cloud.awsimportGlueAdapterfromquerymesh.configimportDatabaseConfigadapter=GlueAdapter(
config=DatabaseConfig(database="my_glue_db"),
s3_staging_dir="s3://my-bucket/athena-results/",
region="us-east-1",
)
adapter.connect()
rows=adapter.fetch_all("SELECT * FROM my_table LIMIT 100")

Azure Active Directory

fromquerymesh.cloud.azureimportAzureADAdapter, AZURE_SQL_SCOPEfromquerymesh.configimportDatabaseConfigadapter=AzureADAdapter(
base_config=DatabaseConfig(
host="myserver.database.windows.net",
database="mydb",
),
scope=AZURE_SQL_SCOPE, # or AZURE_POSTGRES_SCOPE / AZURE_MYSQL_SCOPEengine="sqlserver",
)
adapter.connect()

Uses DefaultAzureCredential from azure-identity - works with managed identities, service principals, and local az login.

Google Cloud BigQuery

fromquerymesh.adapters.bigqueryimportBigQueryAdapterfromquerymesh.configimportDatabaseConfigadapter=BigQueryAdapter(
config=DatabaseConfig(database="my_project.my_dataset"),
project="my-gcp-project",
)
adapter.connect()
rows=adapter.fetch_all("SELECT * FROM my_table LIMIT 100")

Visualizers

Capture and render query execution flows, and export analytics dashboards.

Query Flow Tracing

fromquerymeshimportDatabaseClient, EventBusfromquerymesh.vizimportQueryTracer, AsciiRenderer, MermaidRenderer, HtmlFlowRendererbus=EventBus()
tracer=QueryTracer(bus)
client=DatabaseClient(adapter=..., event_bus=bus)
client.fetch_all("SELECT * FROM users") # cache missclient.fetch_all("SELECT * FROM users") # cache hit# ASCII - terminal output, no dependenciesgraph=tracer.latest(1)[0]
print(AsciiRenderer().render(graph))
# ================================================================# QUERY FLOW [3.2ms | MISS | 5 rows]# ================================================================# SELECT * FROM users# ----------------------------------------------------------------# --> Query received# ? Cache check# [M] Cache MISS# >>> Executed on adapter [3.1ms]# [W] Result cached# <-- Returned 5 row(s) [3.2ms]# ================================================================# Mermaid - paste into GitHub, Notion, mermaid.liveprint(MermaidRenderer().render(graph))
# HTML - embed in any webpagehtml=HtmlFlowRenderer().render(graph)
# Helpersslow_graphs=tracer.slow() # only slow-query graphserror_graphs=tracer.errors() # only error graphsall_graphs=tracer.all() # newest firsttracer.clear()

Analytics Dashboard

Generates a self-contained HTML file with interactive charts (requires internet to load Chart.js from CDN):

fromquerymesh.vizimportDashboardExporter, VizConfigexporter=DashboardExporter(
client.analytics,
config=VizConfig(dashboard_top_n=10),
)
exporter.export("querymesh_dashboard.html")
# Open querymesh_dashboard.html in any browser

The dashboard includes:

  • Summary cards: total queries, avg latency, cache hit rate, slow query count
  • Bar chart: top N queries by call count
  • Donut chart: cache hit/miss ratio
  • Bar chart: avg execution time per query
  • Slow query table
  • Top N queries table

Live Server

Requires pip install flask. Serves the dashboard and live flow traces with auto-refresh:

fromquerymesh.vizimportLiveServer, VizConfigserver=LiveServer(
analytics_engine=client.analytics,
tracer=tracer,
config=VizConfig(live_server_port=7070, live_server_refresh_s=5),
)
server.start()
print(f"Dashboard: {server.url}")
# Routes:# GET / -> analytics dashboard (auto-refreshes)# GET /flows -> recent query flows (auto-refreshes)# GET /flows/latest -> latest flow as plain ASCII# GET /api/stats -> analytics stats as JSON# GET /api/flows -> recent flows as JSONserver.stop()

VizConfig

fromquerymesh.vizimportVizConfigVizConfig(
enabled=True,
trace_queries=True,
max_traces=500, # ring buffer sizeflow_format="ascii", # "ascii" | "mermaid" | "html"dashboard_path="querymesh_dashboard.html",
dashboard_top_n=10,
live_server=False,
live_server_host="127.0.0.1",
live_server_port=7070,
live_server_refresh_s=5,
)

Internal Store

querymesh can persist its own operational data directly in your database using a _qm_* schema (prefix is configurable). The store uses a separate connection so its writes are never affected by application transaction rollbacks.

fromquerymesh.storeimportStoreManager, StoreConfigfromquerymesh.adapters.sqliteimportSqliteAdapterfromquerymesh.configimportDatabaseConfigfromquerymesh.hooksimportEventBus# Dedicated connection for the storestore_adapter=SqliteAdapter(DatabaseConfig(database="querymesh_store.db"))
store_adapter.connect()
bus=EventBus()
store=StoreManager(
adapter=store_adapter,
event_bus=bus, # auto-logs via EventBusconfig=StoreConfig(
log_queries=True,
log_slow_queries=True,
log_errors=True,
audit_queries=False, # enable for compliance loggingsnapshot_stats=False,
max_query_log_rows=10_000,
),
)
store.bootstrap() # creates all _qm_* tables

What gets stored

TableContentDefault
_qm_query_logEvery query execution: SQL, duration, cache_hit, row_counton
_qm_slow_queriesSlow queries with threshold infoon
_qm_errorsQuery errors with type and messageon
_qm_audit_logFull audit trail of every operationoff
_qm_stats_snapshotsPeriodic analytics engine snapshots (JSON)off
_qm_policiesRuntime-editable key-value configon

PolicyStore

Policies are persisted in _qm_policies and loaded into a fast in-memory cache. Update a row in the DB and call reload() to pick up changes without restarting.

# Shipped built-in defaultsstore.policies.get("cache_ttl") # 60 (int)store.policies.get("slow_query_threshold_ms") # 500.0 (float)store.policies.get("max_retries") # 3 (int)# Set custom policiesstore.policies.set("cache_ttl", 120)
store.policies.set("feature_flags", {"v2_api": True}, value_type="json")
store.policies.set("allowed_regions", ["us", "eu"], value_type="json",
description="Permitted deployment regions")
# Readttl=store.policies.get("cache_ttl") # 120flags=store.policies.get("feature_flags") # {"v2_api": True}# All at onceall_p=store.policies.all() # dict of coerced valuesrecs=store.policies.records() # list of PolicyRecord (with type metadata)# Removestore.policies.delete("cache_ttl")
# Reload from DB (e.g. after an external update)store.policies.reload()

Supported types: str, int, float, bool, json.

Reading logs

# Query logrecent=store.query_log(limit=50) # newest firsthits=store.query_log(cache_hit=True)
misses=store.query_log(cache_hit=False)
# Slow queries (sorted by execution time, slowest first)slow=store.slow_queries(limit=10)
# Errorserrs=store.errors(limit=20)
foreinerrs:
print(e.error_type, e.error_message, e.query[:60])
# Audit logaudit=store.audit_log(limit=100)
# Stats snapshotssnaps=store.stats_snapshots(limit=5)
print(snaps[0].snapshot["stats"]["total_queries"])

Summary and export

summary=store.summary()
# {# "query_log": {"total": 412, "cache_hits": 330, "avg_execution_ms": 2.4},# "slow_queries": {"total": 8},# "errors": {"total": 2},# "audit_log": {"total": 412},# "stats_snapshots":{"total": 0},# "policies": {"total": 7},# }data=store.export() # full serialisable dict# Manual log writes (without EventBus)store.log_query("SELECT * FROM users", execution_time_ms=5.2, cache_hit=False)
store.log_error("SELECT * FROM broken", error=RuntimeError("table missing"))
store.log_audit("DROP TABLE users", success=False, error_message="permission denied")
# Force prune log tablesdeleted=store.prune() # {"query_log": 0, "errors": 0, "audit_log": 0}store.close()

Audit Stream

AuditStream adds event-sourcing semantics on top of the store: it auto-subscribes to an EventBus, records every query lifecycle event as an AuditRecord, and lets you replay or subscribe to those records.

fromquerymesh.store.audit_streamimportAuditStreamfromquerymesh.hooksimportEventBusbus=EventBus()
client=DatabaseClient(adapter=adapter, event_bus=bus)
# Standalone (in-memory ring buffer, no StoreManager required)stream=AuditStream(bus, buffer_size=10_000)
stream.start()
client.connect()
client.fetch_all("SELECT * FROM users")
client.execute("INSERT INTO orders (total) VALUES (?)", (99.99,))
# Replay all events since an hour agoimporttimeforrecordinstream.replay(since=time.time() -3600):
print(record.operation, record.query[:60], record.success)
# Get the 5 most recent recordslatest=stream.latest(5)
# Live subscription — fires for every new recordstream.subscribe(lambdar: print("LIVE:", r.operation, r.query[:40]))
stream.stop()

With optional persistent storage via a StoreManager — replay from the full audit log in the DB:

stream=AuditStream(bus, store=store_manager)
stream.start()
# replay() reads from the DB when a store is attachedforrecordinstream.replay(since=0.0):
print(record.timestamp, record.operation, record.query[:60])

StoreConfig

fromquerymesh.storeimportStoreConfigStoreConfig(
enabled=True,
log_queries=True,
log_slow_queries=True,
log_errors=True,
audit_queries=False,
snapshot_stats=False,
snapshot_interval_s=300.0,
max_query_log_rows=10_000, # 0 = no pruningmax_error_log_rows=1_000,
max_audit_log_rows=5_000,
schema_prefix="_qm_", # change to avoid naming conflicts
)

Multi-Tenancy

TenantRouter maps tenant IDs to DatabaseClient instances for per-tenant database isolation:

fromquerymesh.tenancyimportTenantRouterfromquerymeshimportDatabaseClientfromquerymesh.adapters.sqliteimportSqliteAdapterfromquerymesh.configimportDatabaseConfigclient_acme=DatabaseClient(SqliteAdapter(DatabaseConfig(database="acme.db")))
client_globex=DatabaseClient(SqliteAdapter(DatabaseConfig(database="globex.db")))
router=TenantRouter(
{"acme": client_acme, "globex": client_globex},
default_tenant="acme",
)
router.connect_all()

Context manager — yields the tenant's client for the duration of the block:

withrouter.use("acme") asclient:
rows=client.fetch_all("SELECT * FROM orders")

Thread-local current tenant — useful in web middleware (set once per request):

# In middleware / request setup:router.set_current(request.tenant_id)
# Anywhere in the same thread:client=router.get_current()
rows=client.fetch_all("SELECT * FROM orders")

Dynamic registration:

router.register("newcorp", DatabaseClient(...))
router.unregister("acme")
print(router.all_tenant_ids()) # ["globex", "newcorp"]print("acme"inrouter) # Falseprint(len(router)) # 2

Lifecycle helpers:

router.connect_all() # connect every registered clientrouter.disconnect_all() # disconnect every registered client# or use as a context manager:withrouter:
... # connect_all on enter, disconnect_all on exit

CLI

querymesh ships with a command-line tool available as python -m querymesh (or querymesh after pip install).

inspect — live schema

python -m querymesh inspect --db sqlite:///app.db
python -m querymesh inspect --db sqlite:///app.db --counts # include row counts

Output:

Engine : sqlite
Database : app.db
Tables : 3
users [1042 rows]
id INTEGER PK NOT NULL
name TEXT NOT NULL
email TEXT NOT NULL
[index] UNIQUE idx_users_email (email)
orders [8310 rows]
...

migrate — apply SQL migrations

python -m querymesh migrate --db sqlite:///app.db --dir ./migrations
python -m querymesh migrate --db sqlite:///app.db --dir ./migrations --rollback

stats — print analytics as JSON

python -m querymesh stats --db sqlite:///app.db
python -m querymesh stats --db sqlite:///app.db --top 20

viz — live analytics dashboard

python -m querymesh viz --db sqlite:///app.db --port 5000
python -m querymesh viz --db sqlite:///app.db --port 8080 --host 0.0.0.0 --refresh 3

Opens a Flask dashboard at http://127.0.0.1:5000/ with auto-refreshing charts, query flow traces, and JSON API endpoints.

Supported DSN formats:

FormatExample
SQLitesqlite:///path/to/app.db
SQLite in-memorysqlite:///:memory:
PostgreSQLpostgresql://user:pass@host:5432/dbname
MySQLmysql://user:pass@host:3306/dbname

Testing Adapters

MockAdapter - returns pre-configured results with no database required:

fromquerymesh.adapters.mockimportMockAdaptermock=MockAdapter({
"SELECT * FROM users": [{"id": 1, "name": "Alice"}],
"SELECT * FROM users WHERE id = ?": {"id": 1, "name": "Alice"},
"SELECT COUNT(*) AS cnt FROM users": {"cnt": 1},
}, raise_on_unknown=True) # raise on unexpected queriesmock.connect()
rows=mock.fetch_all("SELECT * FROM users")

RecordingAdapter - wraps any adapter and records every call:

fromquerymesh.adapters.mockimportRecordingAdapterfromquerymesh.adapters.sqliteimportSqliteAdapterrecorder=RecordingAdapter(SqliteAdapter(DatabaseConfig(database=":memory:")))
recorder.connect()
recorder.execute("INSERT INTO t VALUES (?)", params=(1,))
recorder.fetch_all("SELECT * FROM t")
recorder.calls# [RecordedCall(method='execute', ...), ...]recorder.queries_for("fetch_all") # ["SELECT * FROM t"]recorder.reset_calls()

Configuration Reference

fromquerymesh.configimport (
DatabaseConfig,
CacheConfig,
AnalyticsConfig,
PoolConfig,
RetryConfig,
CircuitBreakerConfig,
)
fromquerymesh.storeimportStoreConfigfromquerymesh.vizimportVizConfigDatabaseConfig(
host="localhost",
port=5432,
user="app",
password="secret",
database="mydb",
connect_timeout=10,
extra={}, # driver-specific kwargs
)
CacheConfig(
enabled=True,
backend="memory", # "memory" | "redis"ttl=300, # seconds; None = no expirymax_size=1000, # only for memory backendcache_select_only=True,
redis_url="redis://localhost:6379/0",
)
AnalyticsConfig(
enabled=True,
slow_query_threshold_ms=500.0,
)
PoolConfig(
min_size=1,
max_size=10,
checkout_timeout=30.0,
max_idle_seconds=300,
)
RetryConfig(
max_attempts=3,
base_backoff_ms=100.0,
backoff_multiplier=2.0,
max_backoff_ms=5000.0,
jitter=True,
)
CircuitBreakerConfig(
failure_threshold=5,
recovery_timeout_s=60.0,
success_threshold=2,
)
StoreConfig(
enabled=True,
log_queries=True,
log_slow_queries=True,
log_errors=True,
audit_queries=False,
snapshot_stats=False,
snapshot_interval_s=300.0,
max_query_log_rows=10_000,
max_error_log_rows=1_000,
max_audit_log_rows=5_000,
schema_prefix="_qm_",
)
VizConfig(
enabled=True,
trace_queries=True,
max_traces=500,
flow_format="ascii",
dashboard_path="querymesh_dashboard.html",
dashboard_top_n=10,
live_server=False,
live_server_host="127.0.0.1",
live_server_port=7070,
live_server_refresh_s=5,
)

Adding a Custom Adapter

Subclass DBAdapter and implement the six required methods:

fromquerymesh.interfaces.adapterimportDBAdapterfromquerymesh.configimportDatabaseConfigclassMyAdapter(DBAdapter):
def__init__(self, config: DatabaseConfig) ->None:
self._config=configself._connection=Nonedefconnect(self) ->None:
self._connection=my_driver.connect(...)
defdisconnect(self) ->None:
ifself._connection:
self._connection.close()
self._connection=Nonedefexecute(self, query: str, params=None) ->None:
cursor=self._connection.cursor()
cursor.execute(query, paramsor ())
self._connection.commit()
deffetch_one(self, query: str, params=None):
cursor=self._connection.cursor()
cursor.execute(query, paramsor ())
row=cursor.fetchone()
ifrowisNone:
returnNonecols= [d[0] fordincursor.description]
returndict(zip(cols, row))
deffetch_all(self, query: str, params=None):
cursor=self._connection.cursor()
cursor.execute(query, paramsor ())
rows=cursor.fetchall()
cols= [d[0] fordincursor.description]
return [dict(zip(cols, row)) forrowinrows]
defis_connected(self) ->bool:
returnself._connectionisnotNone@propertydefdb_context(self) ->str:
returnf"mydb://{self._config.host}/{self._config.database}"

Optional overrides: execute_many(), begin_transaction(), commit_transaction(), rollback_transaction().

Running Tests

pip install pytest pytest-asyncio aiosqlite
pytest tests/

The test suite requires no external services. All tests run against in-memory SQLite or MockAdapter. Async tests require aiosqlite.

# Run the standalone smoke test (covers all 21 feature areas locally)
python smoke_test.py
183 checks, 183 passed

About

Querymesh: A universal Python database library with smart query caching and analytics.

querymesh.org

Topics

cachedatabase-managementmysqlsql

Resources

Readme
MIT license
Activity

Stars

1 star

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages