# Aurora DSQL dialect for SQLAlchemy
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy"></a>

The Amazon Aurora DSQL dialect for SQLAlchemy (`aurora-dsql-sqlalchemy`) integrates SQLAlchemy object-relational mapping (ORM) with Aurora DSQL. It handles IAM authentication, Secure Sockets Layer (SSL) configuration, DDL generation differences (async indexes, identity columns), and connection management, so you can use the full SQLAlchemy API against a distributed Aurora DSQL cluster with minimal configuration. For more information, see the [Aurora DSQL dialect for SQLAlchemy](https://github.com/awslabs/aurora-dsql-orms/tree/main/python/sqlalchemy) and the [SQLAlchemy pet clinic example application](https://github.com/awslabs/aurora-dsql-orms/tree/main/python/sqlalchemy/examples/pet-clinic-app) on the GitHub website.

This topic contains the following sections:
+ [Prerequisites](#SECTION_program-with-dsql-dialect-for-sqlalchemy-prerequisites)
+ [Installation](#SECTION_program-with-dsql-dialect-for-sqlalchemy-installation)
+ [Quick start](#SECTION_program-with-dsql-dialect-for-sqlalchemy-quick-start)
+ [Configuration](#SECTION_program-with-dsql-dialect-for-sqlalchemy-configuration)
+ [Basic usage](#SECTION_program-with-dsql-dialect-for-sqlalchemy-basic-usage)
+ [Primary key generation](#SECTION_program-with-dsql-dialect-for-sqlalchemy-primary-key-generation)
+ [Index options](#SECTION_program-with-dsql-dialect-for-sqlalchemy-index-options)
+ [OCC retry pattern](#SECTION_program-with-dsql-dialect-for-sqlalchemy-occ-retry-pattern)
+ [SSL configuration](#SECTION_program-with-dsql-dialect-for-sqlalchemy-ssl-configuration)
+ [Connection pooling](#SECTION_program-with-dsql-dialect-for-sqlalchemy-connection-pooling)
+ [Best practices](#SECTION_program-with-dsql-dialect-for-sqlalchemy-best-practices)
+ [Dialect features reference](#SECTION_program-with-dsql-dialect-for-sqlalchemy-dialect-features-reference)

## About the dialect
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-about-the-dialect"></a>

Amazon Aurora DSQL is a serverless, distributed relational database service optimized for transactional workloads. Aurora DSQL has specific requirements, including IAM-based authentication, asynchronous index creation, DDL transaction restrictions, and foreign key enforcement through optimistic concurrency control (OCC), where conflicts raise a serialization error that your application can retry.

With the Aurora DSQL dialect for SQLAlchemy, you can use the full SQLAlchemy ORM API against Aurora DSQL without manually accounting for its behavioral differences. The dialect handles these differences automatically.

### What is Aurora DSQL authentication?
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-what-is-aurora-dsql-authentication"></a>

In Aurora DSQL, authentication involves:
+ **IAM authentication:** All connections use IAM-based authentication with time-limited tokens.
+ **Token generation:** The connector generates authentication tokens from your AWS credentials. Tokens have configurable lifetimes.

The Aurora DSQL dialect for SQLAlchemy works with the `aurora-dsql-python-connector`, which automatically generates IAM authentication tokens when establishing connections.

### Features
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-features"></a>

The dialect provides the following features:
+ **Automatic IAM authentication** – Generates IAM tokens automatically from your AWS credentials.
+ **SSL** – Defaults to `verify-full` mode with the system certificate authority (CA) trust store.
+ **Foreign key support** – The dialect supports `FOREIGN KEY` constraints. Conflicts raise a serialization error (`OC000` / `SQLSTATE 40001`), so your application needs retry logic. For the foreign key options that Aurora DSQL supports, see [Working with foreign key constraints in Aurora DSQL](working-with-foreign-key-constraints.md).
+ **Async index creation** – Creates all indexes with `CREATE INDEX ASYNC`.
+ **Identity columns and sequences** – `autoincrement=True` generates `GENERATED BY DEFAULT AS IDENTITY (CACHE 65536)`; `Identity(always=True, cache=65536)` generates `GENERATED ALWAYS AS IDENTITY (CACHE 65536)`.
+ **UUID primary keys** – Supports server-generated universally unique identifiers (UUIDs) through `gen_random_uuid()`.
+ **Index extensions** – Supports the `INCLUDE` clause and the `NULLS NOT DISTINCT` index option.
+ **Driver support** – Compatible with psycopg (v3, default and recommended), psycopg2, and asyncpg (through the `aurora-dsql-python-connector` directly).

## Prerequisites
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-prerequisites"></a>

Before you use the dialect, make sure that you have the following:
+ Python 3.10 or higher
+ SQLAlchemy 2.0.42 or higher
+ One of the following PostgreSQL drivers:
  + psycopg 3.2.0 or higher (`psycopg[binary]`) — default and recommended
  + psycopg2 2.9.8 or higher (`psycopg2-binary`)
+ AWS credentials with `dsql:DbConnectAdmin` permission (admin) or `dsql:DbConnect` (non-admin)

**Note**  
The `aurora-dsql-sqlalchemy` dialect supports psycopg and psycopg2. For asyncpg connections, use the `aurora-dsql-python-connector` directly.

## Installation
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-installation"></a>

```
PROMPT> pip install aurora-dsql-sqlalchemy
PROMPT> pip install "aurora-dsql-python-connector[psycopg]>=0.2.7"  # includes psycopg_pool and is_occ_error

# Install a driver (choose one):
PROMPT> pip install "psycopg[binary]"   # psycopg v3 (recommended)
PROMPT> pip install psycopg2-binary     # psycopg2
```

**Note**  
Use `pip install "psycopg[binary]"`. The `psycopg-binary` package is a prebuilt-wheel distribution that isn't intended for direct installation.

## Quick start
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-quick-start"></a>

The following example creates an engine. In this example, replace {{cluster-id}} and {{region}} with the values for your cluster.

```
from aurora_dsql_sqlalchemy import create_dsql_engine

engine = create_dsql_engine(
    host="{{cluster-id}}.dsql.{{region}}.on.aws",
    user="admin",
    driver="psycopg",   # or "psycopg2"
)
```

When you call `create_dsql_engine`, it automatically does the following:
+ Generates a short-lived IAM auth token via `aurora-dsql-python-connector`
+ Configures SSL (`sslmode=verify-full`, `sslrootcert=system`)
+ Enables direct TLS negotiation when libpq >= 17 is available
+ Sets up connection pooling with sensible defaults

## Configuration
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-configuration"></a>

### `create_dsql_engine` parameters
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-create-dsql-engine-parameters"></a>


| Parameter | Type | Default | Description | 
| --- | --- | --- | --- | 
| host | str | required | Aurora DSQL cluster endpoint | 
| user | str | required | Database user (for example, "admin") | 
| driver | str | "psycopg" | "psycopg" (v3, default) or "psycopg2" | 
| dbname | str | "postgres" | Database name | 
| sslmode | str | "verify-full" | SSL verification mode | 
| sslrootcert | str | "system" | CA certificate path or "system" | 
| application\_name | str | "sqlalchemy" | Connection label visible in pg\_stat\_activity | 
| pool\_size | int | 5 | Connections to keep open in the pool | 
| max\_overflow | int | 10 | Extra connections allowed beyond pool\_size | 
| connect\_args | dict | None | Extra args passed to the connector (for example, region, profile, token\_duration\_secs) | 
| \*\*engine\_kwargs | Not applicable | Not applicable | Any additional keyword arguments for create\_engine() | 

### Manual engine creation (advanced)
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-manual-engine-creation-advanced"></a>

If you need full control over the uniform resource locator (URL) and connector, you can create the engine manually. The following example creates a SQLAlchemy engine with the Aurora DSQL connector:

```
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
import aurora_dsql_psycopg as dsql_connector

url = URL.create("auroradsql+psycopg", username="admin",
                 host="{{cluster-id}}.dsql.{{region}}.on.aws", database="postgres")

engine = create_engine(
    url,
    creator=lambda: dsql_connector.DSQLConnection.connect(
        host="{{cluster-id}}.dsql.{{region}}.on.aws",
        user="admin",
        sslmode="verify-full",
        sslrootcert="system",
    ),
    pool_size=5,
    max_overflow=10,
)
```

## Basic usage
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-basic-usage"></a>

The following example defines a model, creates its table, and then inserts and queries a row:

```
from sqlalchemy import String
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from sqlalchemy.sql import text
from aurora_dsql_sqlalchemy import create_dsql_engine

engine = create_dsql_engine(host="{{cluster-id}}.dsql.{{region}}.on.aws", user="admin")

class Base(DeclarativeBase):
    pass

class Owner(Base):
    __tablename__ = "owner"
    id: Mapped[str] = mapped_column(UUID, primary_key=True, default=text("gen_random_uuid()"))
    name: Mapped[str] = mapped_column(String(30))
    city: Mapped[str] = mapped_column(String(80))

# Create tables
Base.metadata.create_all(engine)

# Insert
with Session(engine) as session:
    session.add(Owner(name="Mary Major", city="Seattle"))
    session.commit()

# Query
with Session(engine) as session:
    owners = session.query(Owner).filter_by(city="Seattle").all()
    print(owners)
```

## Primary key generation
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-primary-key-generation"></a>

### UUID (recommended)
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-uuid-recommended"></a>

Server-generated UUIDs are the recommended primary key type for Aurora DSQL.

```
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.sql import text

id: Mapped[str] = mapped_column(
    UUID,
    primary_key=True,
    default=text("gen_random_uuid()")
)
```

### Sequences
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-sequences"></a>

```
from sqlalchemy import BigInteger, Sequence

id: Mapped[int] = mapped_column(
    BigInteger,
    primary_key=True,
    default=Sequence("my_seq", cache=65536)
)
```

Aurora DSQL requires sequence cache to be either 1 or >= 65536. The dialect automatically adds `CACHE 65536` (the minimum allowed non-trivial value) if you don't specify a cache, and upgrades `INTEGER` sequences to `BIGINT`.

### Identity columns
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-identity-columns"></a>

```
from sqlalchemy import BigInteger, Identity

id: Mapped[int] = mapped_column(
    BigInteger,
    Identity(always=True, cache=65536),  # generates: GENERATED ALWAYS AS IDENTITY (CACHE 65536)
    primary_key=True,
)

# Or using autoincrement (dialect adds CACHE 65536 automatically):
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
```

## Index options
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-index-options"></a>

### `INCLUDE` clause
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-include-clause"></a>

```
from sqlalchemy import Index

Index("idx_name", table.c.id, auroradsql_include=["name", "email"])
# → CREATE INDEX ASYNC idx_name ON table (id) INCLUDE (name, email)
```

### `NULLS NOT DISTINCT`
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-nulls-not-distinct"></a>

```
Index("idx_unique_col", table.c.column, unique=True, auroradsql_nulls_not_distinct=True)
# → CREATE UNIQUE INDEX ASYNC idx_unique_col ON table (column) NULLS NOT DISTINCT
```

## OCC retry pattern
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-occ-retry-pattern"></a>

Aurora DSQL uses optimistic concurrency control (OCC). Conflicting concurrent transactions raise a `DBAPIError` with one of two error codes:
+ **`OC000`** — Data conflict: two transactions modified the same row concurrently.
+ **`OC001`** — Schema conflict: the transaction's cached schema catalog is out of date.

Applications should catch both and retry. The following example retries a statement when an OCC conflict occurs:

```
import random
import time

from sqlalchemy import text
from sqlalchemy.exc import DBAPIError
from aurora_dsql_psycopg import is_occ_error  # requires aurora-dsql-python-connector >=0.2.7

def execute_with_retry(engine, statement, max_retries=5):
    with engine.connect() as conn:
        for attempt in range(max_retries):
            try:
                conn.execute(text(statement))
                conn.commit()
                return
            except DBAPIError as e:
                conn.rollback()
                if not is_occ_error(e.orig):
                    raise
                if attempt == max_retries - 1:
                    raise
                time.sleep(random.uniform(0, 0.1 * 2 ** attempt))  # exponential backoff with jitter
```

## SSL configuration
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-ssl-configuration"></a>

Aurora DSQL requires TLS for all connections. The following table describes the supported `sslmode` values.


| `sslmode` value | Behavior | 
| --- | --- | 
| verify-full (default) | Verifies server certificate and hostname — recommended | 
| verify-ca | Verifies certificate only, not hostname | 
| require | Encrypts connection but skips certificate verification | 

`sslrootcert="system"` uses the operating system's CA trust store. On Linux, the `psycopg[binary]` and `psycopg2-binary` wheels can't locate the system trust store. Set the `SSL_CERT_FILE` environment variable to your CA bundle, or pass the bundle path as `sslrootcert`. To pin a specific CA:

```
engine = create_dsql_engine(
    host="{{cluster-id}}.dsql.{{region}}.on.aws",
    user="admin",
    sslrootcert="/path/to/custom-ca.pem",
)
```

**Important**  
The default `sslrootcert="system"` works only with `sslmode="verify-full"`. To use `verify-ca`, set `sslrootcert` to a CA file path. To use `require`, pass `sslrootcert=None` to clear the default:  

```
engine = create_dsql_engine(
    host="{{cluster-id}}.dsql.{{region}}.on.aws",
    user="admin",
    sslmode="require",
    sslrootcert=None,
)
```

## Connection pooling
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-connection-pooling"></a>

`create_dsql_engine` creates a pooled engine by default.

```
engine = create_dsql_engine(
    host="{{cluster-id}}.dsql.{{region}}.on.aws",
    user="admin",
    pool_size=10,       # steady-state connections
    max_overflow=20,    # burst capacity above pool_size
)
```

**Important**  
Aurora DSQL enforces a **one-hour maximum connection duration**. The cluster closes long-lived connections. Set `pool_recycle` below 3600 to avoid race conditions — 3300 provides a safe margin:  

```
engine = create_dsql_engine(..., pool_recycle=3300)  # recycle before the 1-hour limit
```

## Best practices
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-best-practices"></a>
+ Use UUID primary keys with `gen_random_uuid()` to avoid sequence hot-spots in distributed writes.
+ The dialect supports foreign key constraints, including referential actions, match types, and deferrable foreign keys. Enforcement is OCC-based: foreign key (FK) conflicts raise a serialization error (`OC000` / `SQLSTATE 40001`) that you can retry in the same way as general write conflicts. FK constraint violations (referencing a non-existent row) raise a separate exception class and aren't retryable; don't use `is_occ_error` to gate retries on those.
+ Aurora DSQL supports `SET CONSTRAINTS` for foreign keys only. For existing tables, add FK or CHECK constraints using `ALTER TABLE ... ADD CONSTRAINT ... NOT VALID`; the dialect appends `NOT VALID` automatically. Validating existing rows requires a separate async step. Record the `job_id` that `ALTER TABLE ASYNC` returns, then check the `succeeded` value that `sys.wait_for_job` returns:

  ```
  ALTER TABLE t ADD CONSTRAINT fk_name FOREIGN KEY (col) REFERENCES other(id) NOT VALID;
  ALTER TABLE ASYNC t VALIDATE CONSTRAINT fk_name;
  -- Use the job_id returned by the preceding statement:
  CALL sys.wait_for_job('returned_job_id');  -- verify succeeded is true
  ```

  Confirm that `sys.wait_for_job` returns `succeeded` as true. `convalidated` becomes `true` only after the job succeeds.
+ Wrap write transactions in OCC retry loops to handle concurrent conflicts.
+ Set `pool_recycle=3300` to stay within Aurora DSQL's one-hour connection duration limit with a safe margin.
+ Use `INCLUDE` indexes for covering queries to reduce secondary reads.
+ Use `Identity(always=True, cache=65536)` or an explicit `Sequence` with `cache=65536` for integer primary keys. Without an explicit `cache` argument, `Identity(always=True)` generates `GENERATED ALWAYS AS IDENTITY` without a CACHE clause. Aurora DSQL supports sequence cache values of 1 or at least 65536, so include an explicit supported cache value.
+ Issue DDL and DML in separate transactions. Each transaction can contain only one DDL statement.

## Dialect features reference
<a name="SECTION_program-with-dsql-dialect-for-sqlalchemy-dialect-features-reference"></a>


| Capability | Supported | Notes | 
| --- | --- | --- | 
| CREATE INDEX ASYNC | Yes | All indexes use async creation | 
| FOREIGN KEY constraints | Yes | Supports referential actions, MATCH SIMPLE and MATCH FULL, and deferrable constraints. Conflicts raise OC000 (SQLSTATE 40001). For details, see [Best practices](#SECTION_program-with-dsql-dialect-for-sqlalchemy-best-practices). | 
| DDL \+ DML in same transaction | No | Use separate transactions for DDL and DML. Each transaction can contain only one DDL statement. | 
| Sequences | Yes | Sequences must use BIGINT. The dialect adds CACHE 65536. | 
| Identity columns | Yes | autoincrement=True → GENERATED BY DEFAULT AS IDENTITY (CACHE 65536); Identity(always=True, cache=65536) → GENERATED ALWAYS AS IDENTITY (CACHE 65536) | 
| SAVEPOINT | No | The dialect suppresses the init-time SAVEPOINT automatically. | 
| SKIP LOCKED | No | For the supported locking clauses, see [`SELECT` command](working-with-postgresql-compatibility-supported-sql-features.md#dsql-select). | 
| NULLS NOT DISTINCT | Yes | Via auroradsql\_nulls\_not\_distinct index option | 
| INCLUDE clause | Yes | Via auroradsql\_include index option | 
| CHECK constraints | Yes | The dialect appends NOT VALID when you add a CHECK constraint to an existing table. To validate existing rows, see [Best practices](#SECTION_program-with-dsql-dialect-for-sqlalchemy-best-practices). | 
| OCC conflict codes | OC000 (data conflict), OC001 (schema conflict) | Catch DBAPIError and check e.orig |