PostgreSQL example

This application keeps ordinary PostgreSQL at the center of the API while exercising the complete pg adapter surface. Boolean selection changes the projected row shape, nullable filters add ordered parameters, and the join only exists when its projected column is requested.

Area	Demonstrated behavior
Read and write	Dynamic queries, CTEs, cardinality helpers, inserts, updates, deletes, transactions
Reuse and concurrency	Prepared queries, transactional batches, PostgreSQL pipeline mode
Large results and data transfer	Cursor-backed async iteration, COPY import and export
Production controls	Standard Schema validation, cancellation, deadlines, read routing, redacted observation

Define the database

The pinned container initializes this small schema:

CREATE TYPE account_status AS ENUM ('active', 'suspended');

CREATE TABLE users (
  id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  email text NOT NULL UNIQUE,
  status account_status NOT NULL DEFAULT 'active'
);

CREATE TABLE projects (
  id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  owner_id bigint NOT NULL REFERENCES users(id),
  name text NOT NULL,
  budget numeric(14, 2)
);

INSERT INTO users (id, email, status) VALUES
  (1, 'alice@example.com', 'active'),
  (2, 'bob@example.com', 'suspended');

INSERT INTO projects (owner_id, name, budget) VALUES
  (1, 'Compiler', 12500.50);

The application selects the PostgreSQL grammar, live schema provider, generated output, project, and one shared type policy:

import { defineConfig } from "@typed-sql/core";
import { postgres, typePolicy } from "@typed-sql/postgres";
import { pg } from "@typed-sql/postgres/pg";

export const connectionString = () =>
  process.env.DATABASE_URL ?? "postgresql://typed_sql:typed_sql_examples@127.0.0.1:55441/typed_sql_examples";

export default defineConfig({
  dialect: postgres({ typePolicy }),
  schema: {
    file: "generated/db/schema.json",
    provider: pg({ connectionString, schemas: ["public"], typePolicy }),
  },
  outDir: "generated/db",
  projects: ["tsconfig.json"],
  typePolicy,
});

pg is installed by the example application. It is not a dependency of the grammar package.

Compose native SQL

The function composes only typed-sql primitives. Values remain parameters; selected columns, joins, and predicates remain explicit structural fragments.

import type { QueryParameters, QueryRow } from "@typed-sql/core";
import { sql } from "@typed-sql/postgres";

export interface AccountFilters {
  readonly status?: "active" | "suspended" | null;
  readonly minimumId?: bigint | null;
}

export interface AccountSelect {
  readonly status: boolean;
  readonly projectBudget: boolean;
}

export function accounts<const Select extends AccountSelect>(filters: AccountFilters, select: Select) {
  return sql`
    SELECT
      account.id,
      account.email
      ${select.status ? sql.fragment`, account.status` : sql.empty}
      ${select.projectBudget ? sql.fragment`, project.budget` : sql.empty}
    FROM users AS account
      ${select.projectBudget ? sql.fragment`LEFT JOIN projects AS project ON project.owner_id = account.id` : sql.empty}
    WHERE 1 = 1
      ${filters.status == null ? sql.empty : sql.fragment`AND account.status = ${filters.status}`}
      ${filters.minimumId == null ? sql.empty : sql.fragment`AND account.id >= ${filters.minimumId}`}
    ORDER BY account.id
  `;
}

export const activeAccounts = accounts({ status: "active", minimumId: 1n }, { status: true, projectBudget: true });

export const accountById = (accountId: bigint) => sql`
  SELECT account.id, account.email, account.status
  FROM users AS account
  WHERE account.id = ${accountId}
`;

export const projectsByOwner = (ownerId: bigint) => sql`
  SELECT project.id, project.owner_id, project.name, project.budget
  FROM projects AS project
  WHERE project.owner_id = ${ownerId}
  ORDER BY project.id
`;

export const accountProjectSummary = sql`
  WITH project_totals AS (
    SELECT
      project.owner_id,
      COUNT(*) AS project_count,
      SUM(project.budget) AS total_budget
    FROM projects AS project
    GROUP BY project.owner_id
  )
  SELECT
    account.id,
    account.email,
    account.status,
    project_totals.project_count,
    project_totals.total_budget
  FROM users AS account
  LEFT JOIN project_totals ON project_totals.owner_id = account.id
  ORDER BY account.id
`;

export type ActiveAccount = QueryRow<typeof activeAccounts>;
export type ActiveAccountParameters = QueryParameters<typeof activeAccounts>;

Hover activeAccounts in a configured editor to see its selected row and ordered parameter tuple. Changing projectBudget or status changes the complete query type instead of widening it to a hand-written interface.

Mutations and cardinality

Mutations are ordinary tagged SQL and use PostgreSQL RETURNING when the caller needs a typed row:

import { sql } from "@typed-sql/postgres";

export interface NewAccount {
  readonly id: bigint;
  readonly email: string;
  readonly status: "active" | "suspended";
}

export interface NewProject {
  readonly id: bigint;
  readonly ownerId: bigint;
  readonly name: string;
  readonly budget: string | null;
}

export const insertAccount = (account: NewAccount) => sql`
  INSERT INTO users (id, email, status)
  VALUES (${account.id}, ${account.email}, ${account.status}::account_status)
  RETURNING id, email, status
`;

export const updateAccountStatus = (accountId: bigint, status: NewAccount["status"]) => sql`
  UPDATE users
  SET status = ${status}::account_status
  WHERE id = ${accountId}
  RETURNING id, email, status
`;

export const deleteAccount = (accountId: bigint) => sql`
  DELETE FROM users
  WHERE id = ${accountId}
  RETURNING id, email, status
`;

export const insertProject = (project: NewProject) => sql`
  INSERT INTO projects (id, owner_id, name, budget)
  VALUES (${project.id}, ${project.ownerId}, ${project.name}, ${project.budget})
  RETURNING id, owner_id, name, budget
`;

export const deleteProjectsByOwner = (ownerId: bigint) => sql`
  DELETE FROM projects
  WHERE owner_id = ${ownerId}
`;

/** The plain, one-row INSERT shape required by PostgreSQL COPY FROM. */
export const bulkAccountInsert = (account: NewAccount) => sql`
  INSERT INTO users (id, email, status)
  VALUES (${account.id}, ${account.email}, ${account.status})
`;

The database contract makes expected cardinality explicit instead of returning the same array shape for every operation:

import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { accountById, activeAccounts } from "./queries.js";

export const listActiveAccounts = (database: PostgresDatabase) => database.all(activeAccounts);

export const requireAccount = (database: PostgresDatabase, accountId: bigint) => database.one(accountById(accountId));

export const findAccount = (database: PostgresDatabase, accountId: bigint) => database.maybeOne(accountById(accountId));

Transactions keep the same inferred query rows across the callback boundary:

import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { insertAccount, insertProject, type NewAccount, type NewProject } from "./mutations.js";

export function createAccountWithProject(
  database: PostgresDatabase,
  account: NewAccount,
  project: Omit<NewProject, "ownerId">,
) {
  return database.transaction(async (transaction) => {
    const insertedAccount = await transaction.one(insertAccount(account));
    const insertedProject = await transaction.one(insertProject({ ...project, ownerId: account.id }));
    return { account: insertedAccount, project: insertedProject };
  });
}

Prepared, batched, and pipelined work

Prepared query factories keep parameter inference at the call site:

import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { accountById, projectsByOwner } from "./queries.js";

export function prepareAccountQueries(database: PostgresDatabase) {
  return Object.freeze({
    accountById: database.prepare("example-account-by-id", accountById),
    projectsByOwner: database.prepare("example-projects-by-owner", projectsByOwner),
  });
}

Independent or transactional work uses adapter capabilities rather than a second query language:

import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { accountById, projectsByOwner } from "./queries.js";

export async function loadAccountWorkspace(database: PostgresDatabase, accountId: bigint) {
  const [accounts, projects] = await database.batch([accountById(accountId), projectsByOwner(accountId)]);
  return { account: accounts[0], projects };
}

import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { accountById, accountProjectSummary } from "./queries.js";

export async function loadIndependentReports(database: PostgresDatabase, accountId: bigint) {
  const [account, summary] = await database.pipeline([accountById(accountId), accountProjectSummary]);
  return { account: account[0], summary };
}

Streaming and COPY

The async iterable is cursor-backed, and collected rows retain QueryRow<typeof activeAccounts>:

import type { QueryRow } from "@typed-sql/core";
import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { activeAccounts } from "./queries.js";

export function streamActiveAccounts(database: PostgresDatabase, batchSize = 100) {
  return database.stream(activeAccounts, { batchSize });
}

export async function collectActiveAccounts(database: PostgresDatabase) {
  const accounts: QueryRow<typeof activeAccounts>[] = [];
  for await (const account of streamActiveAccounts(database, 25)) accounts.push(account);
  return accounts;
}

export async function firstActiveAccount(database: PostgresDatabase) {
  for await (const account of streamActiveAccounts(database, 1)) return account;
  return undefined;
}

COPY is exposed by the PostgreSQL adapter as an explicit capability:

import { requireAdapterCapability } from "@typed-sql/core";
import { postgresCopy, sql } from "@typed-sql/postgres";
import type { PostgresDatabase } from "@typed-sql/postgres/runtime";
import { bulkAccountInsert, type NewAccount } from "./mutations.js";

export function importAccounts(database: PostgresDatabase, accounts: Iterable<NewAccount> | AsyncIterable<NewAccount>) {
  const copy = requireAdapterCapability(database, postgresCopy);
  return copy.copyFrom(bulkAccountInsert, accounts, { chunkBytes: 64 * 1024 });
}

export function exportAccountsAsCsv(database: PostgresDatabase) {
  const copy = requireAdapterCapability(database, postgresCopy);
  return copy.copyTo(sql`
    SELECT id, email, status
    FROM users
    ORDER BY id
  `);
}

Validation and production controls

Any Standard Schema-compatible validator can decode the inferred result before it reaches the caller:

import { sql } from "@typed-sql/postgres";
import * as v from "valibot";

export const accountResult = v.object({
  id: v.bigint(),
  email: v.pipe(v.string(), v.email()),
  status: v.picklist(["active", "suspended"]),
});

interface ValidatedAccount {
  readonly id: bigint;
  readonly email: string;
  readonly status: "active" | "suspended";
}

export const validatedAccountById = (accountId: bigint) =>
  sql.validateResult(
    sql<ValidatedAccount>`
      SELECT account.id, account.email, account.status
      FROM users AS account
      WHERE account.id = ${accountId}
    `,
    accountResult,
  );

Cancellation and deadlines operate on driver work, routing uses a validated grammar snapshot, and observation never exposes query text or parameter values:

import { sql } from "@typed-sql/postgres";
import type { PostgresDatabase } from "@typed-sql/postgres/runtime";

const wait = (seconds: number) => sql`SELECT pg_sleep(${seconds})`;

export function waitUntilDeadline(database: PostgresDatabase, seconds: number, timeoutMilliseconds: number) {
  return database.all(wait(seconds), { deadline: Date.now() + timeoutMilliseconds });
}

export function waitUntilAborted(database: PostgresDatabase, seconds: number, signal: AbortSignal) {
  return database.all(wait(seconds), { signal });
}

import type { Database } from "@typed-sql/core";
import { createPostgresRoutedDatabase, postgres, typePolicy } from "@typed-sql/postgres";

export function createAccountRouter(primary: Database, replicas: readonly Database[], snapshot: unknown) {
  return createPostgresRoutedDatabase({
    primary,
    replicas,
    schema: postgres({ typePolicy }).validateSnapshot(snapshot),
    typePolicy,
  });
}

import type { DatabaseObserver, DatabaseOperationEnd, DatabaseOperationStart } from "@typed-sql/core";

export interface OperationLog {
  readonly starts: readonly DatabaseOperationStart[];
  readonly ends: readonly DatabaseOperationEnd[];
  readonly observer: DatabaseObserver;
}

export function createOperationLog(): OperationLog {
  const starts: DatabaseOperationStart[] = [];
  const ends: DatabaseOperationEnd[] = [];
  return {
    starts,
    ends,
    observer: {
      start(operation) {
        starts.push(operation);
        return { end: (completion) => ends.push(completion) };
      },
    },
  };
}

Execute through pg

The adapter consumes the same query object and closes its application-owned pool:

import { typePolicy } from "@typed-sql/postgres";
import { createPgDatabase } from "@typed-sql/postgres/pg";
import { connectionString } from "../typed-sql.config.js";
import { loadAccountWorkspace } from "./batches.js";
import { findAccount } from "./cardinality.js";
import { createOperationLog } from "./observation.js";
import { loadIndependentReports } from "./pipelines.js";
import { prepareAccountQueries } from "./prepared.js";
import { collectActiveAccounts } from "./streams.js";

const pgCursorPackage: string = "pg-cursor";
const pgCopyStreamsPackage: string = "pg-copy-streams";

export async function runPostgresExample() {
  const operationLog = createOperationLog();
  const database = await createPgDatabase({
    connectionString,
    typePolicy,
    observer: operationLog.observer,
    poolConfig: { pipeline: true },
    cursorImporter: () => import(pgCursorPackage),
    copyStreamsImporter: () => import(pgCopyStreamsPackage),
  });

  try {
    const prepared = prepareAccountQueries(database);
    const [account, preparedAccount, workspace, reports, streamedAccounts] = await Promise.all([
      findAccount(database, 1n),
      database.maybeOne(prepared.accountById(2n)),
      loadAccountWorkspace(database, 1n),
      loadIndependentReports(database, 1n),
      collectActiveAccounts(database),
    ]);
    return {
      account,
      preparedAccount,
      workspace,
      reports,
      streamedAccounts,
      observedOperations: operationLog.ends.length,
    };
  } finally {
    await database.close();
  }
}

The real Poku suite creates actual rows and verifies queries, CTEs, prepared statements, transactions, pipeline mode, cursor streaming, COPY, cancellation, routing, observation, and cleanup against the pinned server:

import { readFile } from "node:fs/promises";
import { QueryCancelledError } from "@typed-sql/core";
import { postgres, sql, typePolicy } from "@typed-sql/postgres";
import { createPgDatabase } from "@typed-sql/postgres/pg";
import { describe, it, strict } from "poku";
import { loadAccountWorkspace } from "../src/batches.js";
import { exportAccountsAsCsv, importAccounts } from "../src/bulk.js";
import { waitUntilAborted, waitUntilDeadline } from "../src/cancellation.js";
import { findAccount, listActiveAccounts, requireAccount } from "../src/cardinality.js";
import { deleteAccount, deleteProjectsByOwner, updateAccountStatus } from "../src/mutations.js";
import { createOperationLog } from "../src/observation.js";
import { loadIndependentReports } from "../src/pipelines.js";
import { prepareAccountQueries } from "../src/prepared.js";
import { accountById, accountProjectSummary } from "../src/queries.js";
import { createAccountRouter } from "../src/routing.js";
import { collectActiveAccounts, firstActiveAccount } from "../src/streams.js";
import { createAccountWithProject } from "../src/transactions.js";
import { validatedAccountById } from "../src/validation.js";
import { connectionString } from "../typed-sql.config.js";

const pgCursorPackage: string = "pg-cursor";
const pgCopyStreamsPackage: string = "pg-copy-streams";
const operationLog = createOperationLog();
const database = await createPgDatabase({
  connectionString,
  typePolicy,
  observer: operationLog.observer,
  poolConfig: { pipeline: true },
  cursorImporter: () => import(pgCursorPackage),
  copyStreamsImporter: () => import(pgCopyStreamsPackage),
});

async function removeAccounts(ids: readonly bigint[]): Promise<void> {
  for (const id of ids) {
    await database.execute(deleteProjectsByOwner(id));
    await database.execute(deleteAccount(id));
  }
}

try {
  await describe("PostgreSQL example against pg", async () => {
    await it("executes queries, cardinality, CTEs, prepared statements, batches, pipelines, and streams", async () => {
      strict.strictEqual((await listActiveAccounts(database)).length, 1);
      strict.deepStrictEqual(await requireAccount(database, 1n), {
        id: 1n,
        email: "alice@example.com",
        status: "active",
      });
      strict.strictEqual(await findAccount(database, -1n), undefined);
      strict.strictEqual((await database.all(accountProjectSummary)).length, 2);

      const prepared = prepareAccountQueries(database);
      strict.strictEqual(prepared.accountById.statementName, "example-account-by-id");
      strict.deepStrictEqual(await database.one(prepared.accountById(2n)), {
        id: 2n,
        email: "bob@example.com",
        status: "suspended",
      });

      const workspace = await loadAccountWorkspace(database, 1n);
      strict.strictEqual(workspace.projects.length, 1);
      const reports = await loadIndependentReports(database, 1n);
      strict.strictEqual(reports.summary.length, 2);
      strict.strictEqual((await collectActiveAccounts(database)).length, 1);
      strict.deepStrictEqual(await firstActiveAccount(database), {
        id: 1n,
        email: "alice@example.com",
        status: "active",
        budget: "12500.50",
      });
      strict.deepStrictEqual(await database.one(validatedAccountById(1n)), {
        id: 1n,
        email: "alice@example.com",
        status: "active",
      });
    });

    await it("commits typed mutations atomically", async () => {
      await removeAccounts([9_001n]);
      const created = await createAccountWithProject(
        database,
        { id: 9_001n, email: "transaction.pg@example.com", status: "active" },
        { id: 9_001n, name: "PostgreSQL transaction", budget: "42.50" },
      );
      strict.deepStrictEqual(created.account, {
        id: 9_001n,
        email: "transaction.pg@example.com",
        status: "active",
      });
      strict.deepStrictEqual(await database.one(updateAccountStatus(9_001n, "suspended")), {
        id: 9_001n,
        email: "transaction.pg@example.com",
        status: "suspended",
      });
      await removeAccounts([9_001n]);
      strict.strictEqual(await findAccount(database, 9_001n), undefined);
    });

    await it("imports and exports typed rows through COPY", async () => {
      const ids = [9_101n, 9_102n] as const;
      await removeAccounts(ids);
      const result = await importAccounts(database, [
        { id: ids[0], email: "copy-one.pg@example.com", status: "active" },
        { id: ids[1], email: "copy-two.pg@example.com", status: "suspended" },
      ]);
      strict.strictEqual(result.rows, 2);
      strict.ok(result.bytes > 0);
      const chunks: Uint8Array[] = [];
      for await (const chunk of exportAccountsAsCsv(database)) chunks.push(chunk);
      const csv = new TextDecoder().decode(Buffer.concat(chunks));
      strict.match(csv, /copy-one\.pg@example\.com/u);
      strict.match(csv, /copy-two\.pg@example\.com/u);
      await removeAccounts(ids);
    });

    await it("cancels in-flight driver work by deadline and AbortSignal", async () => {
      await strict.rejects(waitUntilDeadline(database, 0.1, 10), (error) => {
        strict.ok(error instanceof QueryCancelledError);
        return true;
      });
      const controller = new AbortController();
      const pending = waitUntilAborted(database, 0.1, controller.signal);
      setTimeout(() => controller.abort(), 10);
      await strict.rejects(pending, (error) => {
        strict.ok(error instanceof QueryCancelledError);
        return true;
      });
      strict.deepStrictEqual(await database.one(sql<{ value: number }>`SELECT 1 AS value`), { value: 1 });
    });

    await it("routes proven reads with the grammar snapshot and emits redacted observations", async () => {
      const snapshot = postgres().validateSnapshot(
        JSON.parse(await readFile(new URL("../generated/db/schema.json", import.meta.url), "utf8")),
      );
      const replica = await createPgDatabase({ connectionString, typePolicy, observer: operationLog.observer });
      try {
        const routed = createAccountRouter(database, [replica], snapshot).context();
        strict.deepStrictEqual(await routed.one(accountById(1n)), {
          id: 1n,
          email: "alice@example.com",
          status: "active",
        });
      } finally {
        await replica.close();
      }
      strict.ok(operationLog.starts.length > 0);
      strict.strictEqual(operationLog.starts.length, operationLog.ends.length);
      strict.ok(operationLog.starts.every((event) => !("text" in event) && !("values" in event)));
    });
  });
} finally {
  await removeAccounts([9_001n, 9_101n, 9_102n]).catch(() => undefined);
  await database.close();
}

Run it

From the repository root:

pnpm --filter @typed-sql/example-postgres db:up
pnpm --filter @typed-sql/example-postgres generate
pnpm --filter @typed-sql/example-postgres check
pnpm --filter @typed-sql/example-postgres test
pnpm --filter @typed-sql/example-postgres start
pnpm --filter @typed-sql/example-postgres test:database
pnpm --filter @typed-sql/example-postgres db:down

Set TYPED_SQL_CONTAINER_ENGINE=podman before the container commands to use Podman. Set DATABASE_URL to target another PostgreSQL database. The PostgreSQL dialect guide covers the full grammar and adapter surface.