pymnemon

Unified SQLAlchemy connector with normalized error handling across multiple databases.

Install and import are both pymnemon.

Upgrading from 0.1.x: the import name was mnemon and is now pymnemon. Replace from mnemon import ... with from pymnemon import ....

Install

pip install pymnemon

Quickstart

from pymnemon import SchemaDBConnection

config = {
    "database_type": "postgresql",
    "auth_method": "password",
    "connection": {
        "user": "dbuser",
        "password": "dbpass",
        "host": "localhost",
        "port": 5432,
        "database": "analytics",
        # optional
        # "sslmode": "require",
    },
}

with SchemaDBConnection(config) as db:
    ok, engine, error = db.safe_connect()
    if not ok:
        print(error)
    else:
        rows = db.execute_query("SELECT 1 AS ok")
        print(rows)

Configuration

SchemaDBConnection expects a config dict with this shape:

{
  "database_type": "<supported database>",
  "auth_method": "<supported auth method>",
  "connection": { ... database-specific fields ... }
}

Supported databases and auth methods:

• postgresql: password, scram, ssl_verify, ssl_cert
• mysql: password, ssl_verify, ssl_cert
• mariadb: password, ssl_verify, ssl_cert
• clickhouse: password, ssl_verify, ssl_cert
• sqlserver: password
• trino: none, password, jwt, certificate
• sparksql: none, password, ldap
• vertica: password, ldap, ssl_verify
• oracle: password, wallet
• teradata: password, ldap
• db2: password, ldap
• snowflake: password, key_pair
• bigquery: service_account
• redshift: password, iam_role
• duckdb: local_file, motherduck
• databricks: token, oauth_m2m
• athena: iam_credentials

Example configs

PostgreSQL (password):

{
  "database_type": "postgresql",
  "auth_method": "password",
  "connection": {
    "user": "dbuser",
    "password": "dbpass",
    "host": "localhost",
    "port": 5432,
    "database": "analytics"
  }
}

Snowflake (key pair):

{
  "database_type": "snowflake",
  "auth_method": "key_pair",
  "connection": {
    "account": "xy12345.us-east-1",
    "user": "DBUSER",
    "warehouse": "COMPUTE_WH",
    "database": "ANALYTICS",
    "schema": "PUBLIC",
    "private_key": {"content": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"},
    "private_key_passphrase": "optional"
  }
}

BigQuery (service account):

{
  "database_type": "bigquery",
  "auth_method": "service_account",
  "connection": {
    "project": "my-gcp-project",
    "dataset": "analytics",
    "location": "US",
    "credentials_json": {"content": "{... service account json ...}"}
  }
}

Databricks (PAT):

{
  "database_type": "databricks",
  "auth_method": "token",
  "connection": {
    "host": "adb-1234567890.12.azuredatabricks.net",
    "http_path": "/sql/1.0/warehouses/abcd1234",
    "access_token": "dapi..."
  }
}

Error handling

safe_connect() returns (success, engine, error) where error is a normalized JSON payload with:

{
  "success": false,
  "error": {
    "category": "auth_failed",
    "message": "Authentication failed. Please check your credentials.",
    "next_steps": ["Verify username and password"],
    "field_hint": "user or password",
    "details": "... original error ..."
  }
}

Extending to non-SQL stores

SchemaDBConnection covers 17 SQL dialects. For stores that are not SQL — document, vector, object, filesystem, key-value — pymnemon.store defines a universal store contract: five verbs any store family can implement, plus a thirteen-check conformance suite that verifies an implementation rather than trusting it.

SPEC.md is the normative specification. Read it before writing an adapter.

from pymnemon.store import StoreAdapter, QuerySpec, UnsupportedOperation

class MyAdapter:                       # implements StoreAdapter
    store_id = "my-store"
    kind = "document"

    def connect(self): ...             # -> None | Failure
    def characterize(self, sample_size=200): ...
    def paginate(self, collection, cursor, size): ...
    def query(self, spec: QuerySpec): ...
    def close(self): ...

Verify it:

python -m pymnemon.store mypkg.adapters:build

from pymnemon.store import run_conformance

report = run_conformance(lambda: MyAdapter(config), collection="events")
print(report.render())
assert report.passed

Or as one pytest test per check:

from pymnemon.store.pytest_plugin import conformance_tests

TestMyAdapter = conformance_tests(lambda: MyAdapter(config), collection="events")

pymnemon.store has no dependencies — not SQLAlchemy, not pydantic, not pytest. Writing an adapter for a document store does not require installing eighteen database drivers. pymnemon/store/memory.py is a complete, correct reference implementation to copy from, and doubles as the fixture the suite runs against, so every check is runnable with no live store.

The relational adapter

SQLStoreAdapter implements the contract over any of the 17 dialects, by wrapping SchemaDBConnection:

from pymnemon.store.sql import SQLStoreAdapter

adapter = SQLStoreAdapter(config={"database_type": "postgresql", ...})
# or bring your own engine:
adapter = SQLStoreAdapter(engine=create_engine("sqlite:///data.db"))

adapter.connect()
page = adapter.paginate("events", None, 100)   # keyset, not OFFSET

It uses keyset pagination wherever a primary key exists, merges the catalogue's declared types with an observed sample, and maps driver errors onto the contract's five failure kinds. This is the module to read for a worked example against a real store — and the second implementation is what shows the contract generalizes rather than merely describing MemoryStore.

Importing pymnemon.store.sql requires SQLAlchemy; importing pymnemon.store does not.

The document adapter

MongoStoreAdapter implements the same contract over MongoDB — no catalogue, heterogeneous documents, keyset pagination on _id:

from pymnemon.store.mongo import MongoStoreAdapter

adapter = MongoStoreAdapter(uri="mongodb://localhost:27017/", database="app")

BSON values are normalized on the way out (ObjectId and Decimal128 to str and float, Binary to bytes), so callers never import bson. Requires pymongo.

Verified against

Adapter	Store	Checks
MemoryStore	in-memory (fixture)	13/13
SQLStoreAdapter	SQLite, PostgreSQL	13/13
MongoStoreAdapter	MongoDB 8.0	13/13

Three implementations across relational and document families pass the same thirteen checks with no adapter-specific cases — and none of them required changing the contract.

The rule the contract exists to enforce:

An adapter must raise UnsupportedOperation for anything it cannot express, and must never silently drop it.

A dropped filter returns well-formed, plausible records that answer a different question than the one asked — and nothing downstream can detect it.

Notes on dependencies

Some drivers require system dependencies or extra setup:

• pyodbc for SQL Server requires an ODBC driver (e.g., ODBC Driver 18).
• oracledb may require Oracle client configuration depending on mode.
• PyAthena[SQLAlchemy] is used for Athena SQLAlchemy dialect support.

License

MIT