pymnemon 0.1.1

Unified db connector system with normalized error handling across multiple databases.

pymnemon

Unified SQLAlchemy connector with normalized error handling across multiple databases.

Note: The PyPI package name is pymnemon, but the import name is mnemon.

Install

pip install pymnemon

Quickstart

from mnemon 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 ..."
  }
}

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