DSQL

import { App } from '@anvil-cloud/sdk';
import * as anvil from '@anvil-cloud/sdk';

export default new App({
  run(ctx) {
    const db = new anvil.aws.DSQL('db', {
      schemas: [{
        name: 'app',
        tables: [{
          name: 'users',
          columns: [
            { name: 'id', type: 'uuid' },
            { name: 'email', type: 'text', notNull: true },
          ],
          primaryKey: ['id'],
        }],
        roles: [{ name: 'app_role', grants: ['SELECT', 'INSERT'] }],
      }],
    });

    const fn = new anvil.aws.Lambda('processor', {
      runtime: 'nodejs24.x',
      handler: 'functions/processor/index.handler',
      environment: {
        DSQL_ENDPOINT: db.endpoint,
      },
    });

    // Maps the Lambda's IAM role to the app_role Postgres role (dsql:DbConnect +
    // an AWS IAM GRANT mapping — see the architecture page for what this does).
    db.grantConnect(fn, { schemas: [{ schema: 'app', roleName: 'app_role' }] });

    ctx.export('endpoint', db.endpoint);
  },
});

import anvil_cloud as anvil

def infra(ctx: anvil.Context):
    db = anvil.aws.DSQL("db",
        schemas=[{
            "name": "app",
            "tables": [{
                "name": "users",
                "columns": [
                    {"name": "id", "type": "uuid"},
                    {"name": "email", "type": "text", "not_null": True},
                ],
                "primary_key": ["id"],
            }],
            "roles": [{"name": "app_role", "grants": ["SELECT", "INSERT"]}],
        }],
    )

    fn = anvil.aws.Lambda("processor",
        runtime="nodejs24.x",
        handler="functions/processor/index.handler",
        environment={"DSQL_ENDPOINT": db.endpoint},
    )

    # Maps the Lambda's IAM role to the app_role Postgres role (dsql:DbConnect +
    # an AWS IAM GRANT mapping — see the architecture page for what this does).
    db.grant_connect(fn, schemas=[{"schema": "app", "role_name": "app_role"}])

    ctx.export("endpoint", db.endpoint)

anvil.App(run=infra)

package main

import (
    "github.com/anvil-cloud/anvil/sdk/go/anvil"
    anvilaws "github.com/anvil-cloud/anvil/sdk/go/anvil/aws"
    "github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

func main() {
    anvil.Run(anvil.AppConfig{
        Run: func(ctx *anvil.Context) error {
            db, err := anvilaws.NewDSQL(ctx.PulumiCtx(), "db", &anvilaws.DSQLArgs{
                Schemas: anvilaws.DSQLSchemaArray{
                    anvilaws.DSQLSchemaArgs{
                        Name: pulumi.String("app"),
                        Tables: anvilaws.DSQLTableArray{
                            anvilaws.DSQLTableArgs{
                                Name: pulumi.String("users"),
                                Columns: anvilaws.DSQLColumnArray{
                                    anvilaws.DSQLColumnArgs{Name: pulumi.String("id"), Type: anvilaws.DSQLColumnType("uuid")},
                                    anvilaws.DSQLColumnArgs{Name: pulumi.String("email"), Type: anvilaws.DSQLColumnType("text"), NotNull: pulumi.Bool(true)},
                                },
                                PrimaryKey: pulumi.StringArray{pulumi.String("id")},
                            },
                        },
                        Roles: anvilaws.DSQLRoleArray{
                            anvilaws.DSQLRoleArgs{
                                Name:   pulumi.String("app_role"),
                                Grants: anvilaws.DSQLGrantArray{anvilaws.DSQLGrant("SELECT"), anvilaws.DSQLGrant("INSERT")},
                            },
                        },
                    },
                },
            })
            if err != nil {
                return err
            }

            fn, err := anvilaws.NewLambda(ctx.PulumiCtx(), "processor", &anvilaws.LambdaArgs{
                Runtime: anvilaws.LambdaRuntime_Nodejs24_x,
                Handler: pulumi.String("functions/processor/index.handler"),
                Environment: pulumi.StringMap{"DSQL_ENDPOINT": db.Endpoint},
            })
            if err != nil {
                return err
            }

            // No grantConnect convenience in Go yet — construct DSQLConnect directly.
            if _, err := anvilaws.NewDSQLConnect(ctx.PulumiCtx(), "db-processor-connect", &anvilaws.DSQLConnectArgs{
                ClusterArns:    db.ClusterArns,
                TargetRoleArn:  fn.RoleArn,
                TargetName:     pulumi.String("processor"),
                DbRoles:        pulumi.StringArray{pulumi.String("app_role")},
                RolesTableName: db.RolesTableName,
            }); err != nil {
                return err
            }

            ctx.Export("endpoint", db.Endpoint)
            return nil
        },
    })
}

// db.ts
import { DsqlSigner } from '@aws-sdk/dsql-signer';
import { Client } from 'pg';

const endpoint = process.env.DSQL_ENDPOINT!;
const region = process.env.AWS_REGION!;

// Tokens are short-lived and only used to open the connection, so make one per connect.
export async function connect(): Promise<Client> {
  const token = await new DsqlSigner({ hostname: endpoint, region }).getDbConnectAuthToken();

  const client = new Client({
    host: endpoint,
    port: 5432,
    user: 'app_role',
    password: token,
    database: 'postgres',
    ssl: { rejectUnauthorized: true },
  });
  await client.connect();
  return client;
}

// Anywhere in your app:
const client = await connect();
await client.query(
  'INSERT INTO app.users (id, email) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING',
  [crypto.randomUUID(), '[email protected]'],
);
const { rows } = await client.query('SELECT id, email FROM app.users');
await client.end();

# db.py
import os
import uuid
import boto3
import psycopg2

ENDPOINT = os.environ["DSQL_ENDPOINT"]
REGION = os.environ["AWS_REGION"]

# Tokens are short-lived and only used to open the connection, so make one per connect.
def connect():
    token = boto3.client("dsql", region_name=REGION).generate_db_connect_auth_token(
        Hostname=ENDPOINT, Region=REGION,
    )
    return psycopg2.connect(
        host=ENDPOINT, port=5432, user="app_role",
        password=token, dbname="postgres", sslmode="require",
    )

# Anywhere in your app:
conn = connect()
with conn, conn.cursor() as cur:
    cur.execute(
        "INSERT INTO app.users (id, email) VALUES (%s, %s) ON CONFLICT (id) DO NOTHING",
        (str(uuid.uuid4()), "[email protected]"),
    )
    cur.execute("SELECT id, email FROM app.users")
    rows = cur.fetchall()
conn.close()

// db.go
package db

import (
    "context"
    "fmt"
    "os"

    "github.com/aws/aws-sdk-go-v2/config"
    dsqlauth "github.com/aws/aws-sdk-go-v2/feature/dsql/auth"
    "github.com/jackc/pgx/v5"
)

var (
    endpoint = os.Getenv("DSQL_ENDPOINT")
    region   = os.Getenv("AWS_REGION")
)

// Connect opens a connection as app_role. Tokens are short-lived and only used
// to open the connection, so make one per connect.
func Connect(ctx context.Context) (*pgx.Conn, error) {
    cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
    if err != nil {
        return nil, err
    }

    // GenerateDbConnectAuthToken, not the Admin variant.
    token, err := dsqlauth.GenerateDbConnectAuthToken(ctx, endpoint, region, cfg.Credentials)
    if err != nil {
        return nil, err
    }

    connConfig, err := pgx.ParseConfig(fmt.Sprintf(
        "host=%s port=5432 user=app_role dbname=postgres sslmode=verify-full", endpoint,
    ))
    if err != nil {
        return nil, err
    }
    connConfig.Password = token

    return pgx.ConnectConfig(ctx, connConfig)
}

// Anywhere in your app:
//
//     conn, err := db.Connect(ctx)
//     if err != nil { ... }
//     defer conn.Close(ctx)
//     rows, err := conn.Query(ctx, "SELECT id, email FROM app.users")