MySQL example

This example deliberately has the same application shape as the PostgreSQL version. The selected grammar owns MySQL syntax, placeholders, semantics, and decoding while the application continues to write ordinary SQL.

Area	Demonstrated behavior
Read and write	Dynamic queries, CTEs, cardinality helpers, inserts, updates, deletes, transactions
Reuse and batching	Prepared queries and transactional batches
Large results and data transfer	mysql2-backed async iteration and LOAD DATA LOCAL INFILE import
Production controls	Standard Schema validation, cancellation, deadlines, read routing, redacted observation

Define the database

The pinned container initializes this schema:

CREATE TABLE users (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  email VARCHAR(255) NOT NULL UNIQUE,
  status ENUM('active', 'suspended') NOT NULL DEFAULT 'active'
);

CREATE TABLE projects (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  owner_id BIGINT UNSIGNED NOT NULL,
  name VARCHAR(120) NOT NULL,
  budget DECIMAL(14, 2),
  CONSTRAINT projects_owner_fk FOREIGN KEY (owner_id) REFERENCES users(id)
);

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 config selects MySQL, mysql2 introspection, generated output, and a shared type policy:

import { defineConfig } from "@typed-sql/core";
import { mysql, typePolicy } from "@typed-sql/mysql";
import { mysql2 } from "@typed-sql/mysql/mysql2";

export const connectionUri = () =>
  process.env.DATABASE_URL ?? "mysql://typed_sql:typed_sql_examples@127.0.0.1:53311/typed_sql_examples";

export default defineConfig({
  dialect: mysql({ typePolicy }),
  schema: {
    file: "generated/db/schema.json",
    provider: mysql2({ connectionUri, schemas: ["typed_sql_examples"], typePolicy }),
  },
  outDir: "generated/db",
  projects: ["tsconfig.json"],
  typePolicy,
});

mysql2 belongs to the example application and is not installed by the grammar package.

Compose native SQL

Conditional fragments control selection, the dependent join, and nullable filters. Ordinary interpolations stay ordered driver values.

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

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>;

The MySQL renderer uses ? placeholders while preserving the same parameter tuple represented by the query.

Mutations and cardinality

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

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})
`;

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

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

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

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

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

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

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

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

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

Transactions preserve typed rows through the mysql2 connection callback:

import type { MySqlDatabase } from "@typed-sql/mysql/runtime";
import { insertAccount, insertProject, type NewAccount, type NewProject } from "./mutations.js";
import { accountById } from "./queries.js";

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

Prepared, batched, streamed, and bulk work

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

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

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

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

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

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

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

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

MySQL's native bulk example uses LOAD DATA LOCAL INFILE. It is intentionally an adapter capability rather than a grammar-neutral promise that every database can implement:

import { requireAdapterCapability } from "@typed-sql/core";
import { mysqlBulk } from "@typed-sql/mysql";
import type { MySqlDatabase } from "@typed-sql/mysql/runtime";
import { bulkAccountInsert, type NewAccount } from "./mutations.js";

export function importAccounts(database: MySqlDatabase, accounts: Iterable<NewAccount> | AsyncIterable<NewAccount>) {
  const bulk = requireAdapterCapability(database, mysqlBulk);
  return bulk.loadData(bulkAccountInsert, accounts, { chunkBytes: 64 * 1024 });
}

Validation and production controls

import { sql } from "@typed-sql/mysql";
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,
  );

import { sql } from "@typed-sql/mysql";
import type { MySqlDatabase } from "@typed-sql/mysql/runtime";

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

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

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

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

export function createAccountRouter(primary: Database, replicas: readonly Database[], snapshot: unknown) {
  return createMySqlRoutedDatabase({
    primary,
    replicas,
    schema: mysql({ 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 mysql2

The official adapter executes the query through the application-owned pool:

import { typePolicy } from "@typed-sql/mysql";
import { createMySql2Database } from "@typed-sql/mysql/mysql2";
import { connectionUri } from "../typed-sql.config.js";
import { loadAccountWorkspace } from "./batches.js";
import { findAccount } from "./cardinality.js";
import { createOperationLog } from "./observation.js";
import { prepareAccountQueries } from "./prepared.js";
import { accountProjectSummary } from "./queries.js";
import { collectActiveAccounts } from "./streams.js";

export async function runMySqlExample() {
  const operationLog = createOperationLog();
  const database = await createMySql2Database({ connectionUri, typePolicy, observer: operationLog.observer });

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

The database Poku suite executes and cleans up every documented capability against the pinned MySQL server:

import { readFile } from "node:fs/promises";
import { QueryCancelledError } from "@typed-sql/core";
import { mysql, sql, typePolicy } from "@typed-sql/mysql";
import { createMySql2Database } from "@typed-sql/mysql/mysql2";
import { describe, it, strict } from "poku";
import { loadAccountWorkspace } from "../src/batches.js";
import { 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 { 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 { connectionUri } from "../typed-sql.config.js";

const operationLog = createOperationLog();
const database = await createMySql2Database({ connectionUri, typePolicy, observer: operationLog.observer });

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("MySQL example against mysql2", async () => {
    await it("executes queries, cardinality, CTEs, prepared statements, batches, 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",
      });
      strict.strictEqual((await loadAccountWorkspace(database, 1n)).projects.length, 1);
      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.mysql@example.com", status: "active" },
        { id: 9_001n, name: "MySQL transaction", budget: "42.50" },
      );
      strict.deepStrictEqual(created.account, {
        id: 9_001n,
        email: "transaction.mysql@example.com",
        status: "active",
      });
      await database.execute(updateAccountStatus(9_001n, "suspended"));
      strict.deepStrictEqual(await requireAccount(database, 9_001n), {
        id: 9_001n,
        email: "transaction.mysql@example.com",
        status: "suspended",
      });
      await removeAccounts([9_001n]);
      strict.strictEqual(await findAccount(database, 9_001n), undefined);
    });

    await it("imports typed rows through LOAD DATA LOCAL INFILE", async () => {
      const ids = [9_101n, 9_102n] as const;
      await removeAccounts(ids);
      const result = await importAccounts(database, [
        { id: ids[0], email: "load-one.mysql@example.com", status: "active" },
        { id: ids[1], email: "load-two.mysql@example.com", status: "suspended" },
      ]);
      strict.strictEqual(result.rows, 2);
      strict.ok(result.bytes > 0);
      strict.deepStrictEqual(await database.one(accountById(ids[1])), {
        id: ids[1],
        email: "load-two.mysql@example.com",
        status: "suspended",
      });
      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`SELECT 1 AS value`), { value: 1n });
    });

    await it("routes proven reads with the grammar snapshot and emits redacted observations", async () => {
      const snapshot = mysql().validateSnapshot(
        JSON.parse(await readFile(new URL("../generated/db/schema.json", import.meta.url), "utf8")),
      );
      const replica = await createMySql2Database({ connectionUri, 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-mysql db:up
pnpm --filter @typed-sql/example-mysql generate
pnpm --filter @typed-sql/example-mysql check
pnpm --filter @typed-sql/example-mysql test
pnpm --filter @typed-sql/example-mysql start
pnpm --filter @typed-sql/example-mysql test:database
pnpm --filter @typed-sql/example-mysql db:down

Set TYPED_SQL_CONTAINER_ENGINE=podman before the container commands to use Podman. Set DATABASE_URL to target another MySQL database. The MySQL dialect guide documents its grammar and adapter boundaries.