python-aurora

name	python-aurora
description	AWS Aurora PostgreSQL and Aurora DSQL Python integration using asyncpg. Use when building Python backends with Aurora PostgreSQL or Aurora DSQL.

AWS Aurora Python Integration

Use asyncpg for queries, boto3 for IAM auth token generation, and the Vercel Python SDK for OIDC token retrieval. On Vercel, AWS credentials are obtained via OIDC federation: call get_vercel_oidc_token() from vercel.oidc.aio, then exchange that token for temporary AWS credentials via sts.assume_role_with_web_identity(). Do not rely on VERCEL_OIDC_TOKEN as an environment variable; it is not available in production runtime.

Guidelines

• Use asyncpg.create_pool() with the IAM token as the password. Async access avoids blocking FastAPI's event loop.
• Authenticate via OIDC: call get_vercel_oidc_token() to get the Vercel OIDC token, pass it to sts.assume_role_with_web_identity() to get temporary credentials, then use those credentials with rds.generate_db_auth_token().
• Use $1, $2 parameter syntax for queries (asyncpg's native format). Never use f-strings for SQL.
• Enable SSL (ssl="require") — IAM auth requires an encrypted connection.
• Wrap route handlers in try/except and return clear error JSON. Unhandled errors get buried in deep ASGI stack traces and could lead you in the wrong direction.

Aurora DSQL Restrictions

If the target is Aurora DSQL (not standard Aurora PostgreSQL), these PostgreSQL features are not available:

• No SERIAL / BIGSERIAL / SMALLSERIAL — use VARCHAR IDs (e.g. nanoid)
• No temporary tables, triggers, sequences, partitions
• No ON UPDATE clause in CREATE TABLE
• Must COMMIT after each DDL statement
• Must use CREATE INDEX ASYNC (no ASC/DESC in index ordering)

Dependencies

• asyncpg: async PostgreSQL driver
• boto3: AWS SDK for OIDC credential exchange and IAM token generation
• vercel: Vercel Python SDK for runtime OIDC token retrieval

Environment Variables

• PGHOST — Aurora cluster endpoint
• PGUSER — database user (default: postgres)
• PGDATABASE — database name (default: postgres)
• AWS_REGION — AWS region
• AWS_ROLE_ARN — IAM role to assume via OIDC

Setup

import os
import asyncpg
import boto3
from vercel.oidc import decode_oidc_payload
from vercel.oidc.aio import get_vercel_oidc_token


async def _get_aws_credentials() -> dict:
    """Exchange Vercel's OIDC token for temporary AWS credentials."""
    token = await get_vercel_oidc_token()
    payload = decode_oidc_payload(token)
    project_id = payload.get("project_id")

    sts = boto3.client("sts", region_name=os.environ["AWS_REGION"])
    resp = sts.assume_role_with_web_identity(
        RoleArn=os.environ["AWS_ROLE_ARN"],
        RoleSessionName=f"aurora-fastapi-{project_id or 'session'}",
        WebIdentityToken=token,
    )
    return resp["Credentials"]


async def generate_auth_token() -> str:
    creds = await _get_aws_credentials()
    client = boto3.client(
        "rds",
        region_name=os.environ["AWS_REGION"],
        aws_access_key_id=creds["AccessKeyId"],
        aws_secret_access_key=creds["SecretAccessKey"],
        aws_session_token=creds["SessionToken"],
    )
    return client.generate_db_auth_token(
        DBHostname=os.environ["PGHOST"],
        Port=5432,
        DBUsername=os.environ.get("PGUSER", "postgres"),
    )


pool: asyncpg.Pool | None = None


async  () -> asyncpg.Pool:
     pool
     pool  :
        pool =  asyncpg.create_pool(
            host=os.environ[],
            port=,
            user=os.environ.get(, ),
            password= generate_auth_token(),
            database=os.environ.get(, ),
            ssl=,
        )
     pool

Usage

# Error handling omitted for brevity — see Guidelines
from fastapi import FastAPI

app = FastAPI()

@app.get("/todos")
async def list_todos():
    p = await get_pool()
    rows = await p.fetch("SELECT * FROM todos ORDER BY created_at DESC")
    return [dict(r) for r in rows]

@app.post("/todos")
async def create_todo(title: str):
    p = await get_pool()
    row = await p.fetchrow("INSERT INTO todos (title) VALUES ($1) RETURNING *", title)
    return dict(row)

@app.patch("/todos/{todo_id}")
async def update_todo(todo_id: int, completed: bool):
    p = await get_pool()
    row = await p.fetchrow("UPDATE todos SET completed = $1 WHERE id = $2 RETURNING *", completed, todo_id)
    return dict(row)

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    p =  get_pool()
     p.execute(, todo_id)