Skip to content

SqlAlchemySupport.md

Latest commit

 

History

History
243 lines (180 loc) · 14.2 KB

SqlAlchemySupport.md

File metadata and controls

243 lines (180 loc) · 14.2 KB
Raw

SQLAlchemy Support

The AWS Advanced Python Wrapper can be used as a PEP 249 DBAPI module with SQLAlchemy's create_engine via the creator= factory pattern, for both PostgreSQL (via psycopg v3) and MySQL (via mysql-connector-python).

Prerequisites

Using the wrapper with SQLAlchemy (PostgreSQL)

from sqlalchemy import create_engine, text

from aws_advanced_python_wrapper import release_resources
from aws_advanced_python_wrapper.psycopg import connect

engine = create_engine(
    "postgresql+aws_wrapper_psycopg://",
    creator=lambda: connect(
        "host=database.cluster-xyz.us-east-1.rds.amazonaws.com "
        "dbname=db user=john password=pwd",
        wrapper_dialect="aurora-pg",
        plugins="failover,host_monitoring_v2",
        cluster_id="pg_sqlalchemy",
    ),
)

try:
    with engine.connect() as conn:
        row = conn.execute(text("SELECT pg_catalog.aurora_db_instance_identifier()")).one()
        print(row)
finally:
    engine.dispose()
    release_resources()

The wrapper's connection options (wrapper_dialect, plugins, etc.) are passed as kwargs to connect. PostgreSQL requires the wrapper's custom dialect (postgresql+aws_wrapper_psycopg://) even with the creator= pattern: SQLAlchemy's stock psycopg dialect calls psycopg.TypeInfo.fetch() during initialization, which isinstance-checks its argument against the real psycopg.Connection and raises TypeError: expected Connection or AsyncConnection, got AwsWrapperConnection on the wrapper proxy. The custom dialect unwraps to the native connection for those calls.

Using the wrapper with SQLAlchemy (MySQL)

from sqlalchemy import create_engine, text

from aws_advanced_python_wrapper import release_resources
from aws_advanced_python_wrapper.mysql_connector import connect

engine = create_engine(
    "mysql+aws_wrapper_mysqlconnector://",
    creator=lambda: connect(
        "host=database.cluster-xyz.us-east-1.rds.amazonaws.com "
        "database=db user=john password=pwd",
        wrapper_dialect="aurora-mysql",
        plugins="failover,host_monitoring_v2",
        use_pure=True,
    ),
)

try:
    with engine.connect() as conn:
        row = conn.execute(text("SELECT @@aurora_server_id")).one()
        print(row)
finally:
    engine.dispose()
    release_resources()

Note — use_pure + IAM authentication: For Aurora MySQL, we recommend use_pure=True because the C extension's is_connected can block indefinitely on network failure. However, the IAM Authentication Plugin is incompatible with use_pure=True (the pure-Python driver truncates passwords at 255 chars; IAM tokens are longer). See the README's "Known Limitations" section for details.

Using the custom SQLAlchemy dialects (URL-based)

The wrapper registers two SQLAlchemy dialects via entry-points so create_engine can be driven by URL alone — no creator= lambda needed. This is the idiomatic path for Alembic, 12-factor DATABASE_URL configs, and framework starters that expect a URL string.

PostgreSQL:

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+aws_wrapper_psycopg://john:pwd@"
    "database.cluster-xyz.us-east-1.rds.amazonaws.com:5432/db"
    "?wrapper_dialect=aurora-pg&wrapper_plugins=failover,host_monitoring_v2"
)

MySQL:

from sqlalchemy import create_engine

engine = create_engine(
    "mysql+aws_wrapper_mysqlconnector://john:pwd@"
    "database.cluster-xyz.us-east-1.rds.amazonaws.com:3306/db"
    "?wrapper_dialect=aurora-mysql&wrapper_plugins=failover&use_pure=True"
)

Naming

The wrapper registers as a driver under SQLAlchemy's existing dialects, following SA's <dialect>+<driver> URL convention (the same shape as stock postgresql+psycopg, mysql+mysqlconnector, mysql+aiomysql):

Engine Sync URL Async URL
PostgreSQL postgresql+aws_wrapper_psycopg:// postgresql+aws_wrapper_psycopg:// (same — see below)
MySQL mysql+aws_wrapper_mysqlconnector:// mysql+aws_wrapper_aiomysql://

This keeps the dialect identity correct (engine.dialect.name == "postgresql" / "mysql"), so dialect-specific type compilation, reserved-word handling, and any third-party if dialect.name == ... checks behave as expected.

PostgreSQL uses one URL for both sync and async. psycopg3 is a single DBAPI that does both, so — exactly like stock postgresql+psycopg — create_engine(...) yields the sync dialect and create_async_engine(...) yields the async dialect from the same postgresql+aws_wrapper_psycopg:// URL. The selection is made by which engine factory you call, not by the URL. MySQL cannot share a URL this way because its sync and async paths are different DBAPIs (mysql-connector-python vs aiomysql), hence the distinct +aws_wrapper_mysqlconnector / +aws_wrapper_aiomysql driver names.

URL parameter wrapper_plugins (not plugins)

SQLAlchemy's create_engine reserves the query-string plugins= key for its own engine-plugin loader and strips it from the URL before the dialect sees it. To pass the wrapper's plugins connection property via URL, spell it wrapper_plugins= — the dialect translates it back to plugins= before calling the wrapper's connect(). In the creator-pattern path (where you call connect() directly in Python), continue to use the normal plugins= kwarg; the wrapper_plugins alias is URL-only.

All other wrapper connection options (wrapper_dialect, plugin-specific parameters like failover_timeout_sec, auth parameters like iam_region, etc.) pass through the URL query string unchanged as kwargs to the underlying wrapper's connect().

Both the creator-pattern (shown above) and the URL-based path remain supported. Use whichever fits your configuration surface.

Error handling

Wrapper errors are classified so SQLAlchemy maps them to the correct sqlalchemy.exc.* subclass:

Wrapper error SQLAlchemy error
AwsConnectError sqlalchemy.exc.OperationalError
FailoverError, FailoverSuccessError, FailoverFailedError, TransactionResolutionUnknownError sqlalchemy.exc.OperationalError
QueryTimeoutError sqlalchemy.exc.OperationalError
ReadWriteSplittingError sqlalchemy.exc.InterfaceError
UnsupportedOperationError sqlalchemy.exc.NotSupportedError
AwsWrapperError (generic) sqlalchemy.exc.DBAPIError

Applications writing SA retry loops can except sqlalchemy.exc.OperationalError and catch failover events naturally. Target-driver exceptions (e.g., psycopg.errors.*, mysql.connector.errors.*) are not remapped and flow through SA's dialect-specific classification unchanged.

Resource cleanup

Two things must be torn down at shutdown, in this order:

  1. engine.dispose() — drains SQLAlchemy's QueuePool and closes all pooled DBAPI connections.
  2. aws_advanced_python_wrapper.release_resources() — tears down the wrapper's own background threads (topology monitor, host monitoring, internal pool cleanup).

They are complementary: engine.dispose() does not reach the wrapper's background machinery, and release_resources() does not close SA's pool.

Plugin Compatibility

Plugins are configured identically to non-SA usage — via the plugins connection property (or wrapper_plugins in the URL) and any plugin-specific options.

Plugin name Plugin Code Supported?
Failover Plugin failover ✓
Failover Plugin v2 failover_v2 ✓
Host Monitoring Plugin host_monitoring_v2 or host_monitoring ✓
IAM Authentication Plugin iam ✓
AWS Secrets Manager Plugin aws_secrets_manager ✓
Federated Authentication Plugin federated_auth ✓
Okta Authentication Plugin okta ✓
Custom Endpoint Plugin custom_endpoint ✓
Aurora Stale DNS Plugin stale_dns ✓
Aurora Connection Tracker Plugin aurora_connection_tracker ✓
Fastest Response Strategy Plugin fastest_response_strategy ✓
Blue/Green Deployment Plugin bg ✓
Limitless Plugin limitless ✓
Read Write Splitting Plugin read_write_splitting ✗
Simple Read Write Splitting Plugin srw ✗

Read/write splitting is not supported with SQLAlchemy: the plugins switch instances by setting read_only on a long-lived connection, and there is currently no routing of SQLAlchemy's execution_options(...readonly=True) to the wrapper's read_only attribute. For read/write workloads, use SQLAlchemy's own session binding — see the official SQLAlchemy documentation on the Session API.

Async usage

Note

Async engines require SQLAlchemy's asyncio support (pip install "sqlalchemy[asyncio]", which includes greenlet), plus the async driver: Psycopg for PostgreSQL, aiomysql for MySQL.

The wrapper exposes a native async path. create_async_engine works with the URL-based dialect:

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from aws_advanced_python_wrapper.aio import release_resources_async


async def main() -> None:
    engine = create_async_engine(
        # Same URL as the sync PG example above: create_async_engine selects
        # the async dialect via the sync dialect's get_async_dialect_cls hook.
        "postgresql+aws_wrapper_psycopg://john:pwd@"
        "database.cluster-xyz.us-east-1.rds.amazonaws.com:5432/db"
        "?wrapper_dialect=aurora-pg&wrapper_plugins=failover,host_monitoring_v2"
    )
    try:
        async with engine.connect() as conn:
            row = await conn.execute(
                text("SELECT pg_catalog.aurora_db_instance_identifier()")
            )
            print(row.scalar_one())
    finally:
        await engine.dispose()
        await release_resources_async()


asyncio.run(main())

The async path:

  • Drives psycopg.AsyncConnection (PG) or aiomysql (MySQL) end-to-end. No greenlet hops through the wrapper's own pipeline — only SA's engine itself uses greenlet to adapt async DBAPI results.
  • Supports the same wrapper plugins via wrapper_plugins: failover, host_monitoring_v2, iam, aws_secrets_manager, plus the minor/observability plugins. The Plugin Compatibility table above applies to the async path as well — read/write splitting is not supported under SQLAlchemy in either mode.

Async MySQL usage (via aiomysql):

async def main() -> None:
    engine = create_async_engine(
        "mysql+aws_wrapper_aiomysql://john:pwd@"
        "database.cluster-xyz.us-east-1.rds.amazonaws.com:3306/db"
        "?wrapper_dialect=aurora-mysql&wrapper_plugins=failover"
    )
    try:
        async with engine.connect() as conn:
            row = await conn.execute(text("SELECT @@aurora_server_id"))
            print(row.scalar_one())
    finally:
        await engine.dispose()
        await release_resources_async()

At shutdown, call engine.dispose() first, then aws_advanced_python_wrapper.aio.release_resources_async() to drain async background tasks, then (optionally) the sync release_resources() to drain any sync-side threads the app may also have spun up.

Limitations (current)

  • asyncmy and asyncpg drivers are not supported. aiomysql covers MySQL async; psycopg covers PG async. asyncmy deferred pending user-facing perf demand on Aurora; asyncpg dropped because it's not PEP 249-compliant (would require a separate DBAPI adapter layer).

See also