All Symbols - db documentation

default

Examples

Basic query with Task semantics

import { DB, createInMemory } from "@anabranch/db";

const users = await DB.withConnection(createInMemory(), (db) =>
  db.query("SELECT * FROM users WHERE active = ?", [true])
).run();

Streaming large result sets with error collection

import { DB, createInMemory } from "@anabranch/db";

const { successes, errors } = await DB.withConnection(createInMemory(), (db) =>
  db.stream("SELECT * FROM large_table")
    .withConcurrency(10)
    .map(row => processRow(row))
    .partition()
).run();

Transactions with automatic rollback on error

import { DB, ConstraintViolation, createInMemory } from "@anabranch/db";

const result = await DB.withConnection(createInMemory(), (db) =>
  db.withTransaction(async (tx) => {
    await tx.execute("INSERT INTO orders (user_id) VALUES (?)", [userId]);
    await tx.execute("UPDATE users SET order_count = order_count + 1 WHERE id = ?", [userId]);
    return tx.query("SELECT last_insert_rowid()");
  })
).recoverWhen(
  (e) => e instanceof ConstraintViolation,
  (e) => ({ id: 0, error: e.message })
).run();

Retry with exponential backoff

import { DB } from "@anabranch/db";
import { createPostgres } from "@anabranch/db-postgres";

const users = await DB.withConnection(createPostgres(), (db) =>
  db.query("SELECT * FROM users")
    .retry({ attempts: 3, delay: (attempt) => 100 * Math.pow(2, attempt) })
).run();

Pub/sub with in-memory connector (swap for createPostgres in production)

import { createInMemory } from "@anabranch/db";

const connector = createInMemory();
const ch = await connector.listen("orders").run();

await connector.notify("orders", JSON.stringify({ id: 1 })).run();

for await (const n of ch.successes()) {
  console.log(n.payload);
}

Classes

c

CloseError(message: string)

Failed to close the database connection.

c

ConnectionFailed(message: string)

Failed to establish a database connection.

c

ConstraintViolation(

message: string

Constraint violation (e.g., unique, foreign key).

c

DB(adapter: DBAdapter)

Database wrapper with Task/Stream semantics.

execute(
sql: string,
params?: unknown[]
): Task<number, QueryFailed | ConstraintViolation>
Execute INSERT/UPDATE/DELETE and return affected row count.
executeBatch(
sql: string,
paramsArray: unknown[][]
): Task<number[], QueryFailed | ConstraintViolation>
Execute multiple commands in a batch. Leverages optimized adapter method if available.
from(adapter: DBAdapter): DB
Wrap an existing adapter in a DB instance.
query<T extends Record<string, any> = Record<string, any>>(
sql: string,
params?: unknown[]
): Task<T[], QueryFailed | ConstraintViolation>
Execute a SELECT query and return rows.
stream<T extends Record<string, any> = Record<string, any>>(
sql: string,
params?: unknown[]
): Source<T, QueryFailed>
Stream rows from a SELECT query for memory-efficient processing.
transaction(): Task<DBTransaction, TransactionFailed>
withConnection<R, E>(
connector: DBConnector,
fn: (db: DB) => Task<R, E>
): Task<R, E | ConnectionFailed>
Execute operations with a connection acquired from the connector. The connection is automatically released after the operation completes, whether successful or failed.
withTransaction<R>(fn: (tx: DBTransaction) => Promisable<R>): Task<R, TransactionFailed | QueryFailed | ConstraintViolation>
Execute a callback within a transaction. Supports nested transactions via SQL savepoints.

c

sql: string | undefined,

message: string

Structured error types for database operations.

c

adapter: DBTransactionAdapter,

rawAdapter: DBAdapter,

Database transaction with Task semantics.

c

ListenFailed(message: string)

Failed to establish or maintain a pub/sub subscription.

c

message: string

Query execution failed.

c

SerializationFailure(message?: string)

Serialization failure (concurrent modification detected).

c

TransactionFailed(message: string)

Transaction failed.

Functions

f

createInMemory():

DBConnector
& { listen(channel: string): Task<Channel<Notification, ListenFailed>, ListenFailed>; notify(

channel: string,

payload: string

): Task<void, ListenFailed>; }

Creates an in-memory SQLite connector for testing.

Interfaces

I

Database adapter interface for DB-agnostic operations.

close(): Promise<void>
Release the connection back to its source (e.g., pool).
execute(
sql: string,
params?: unknown[]
): Promise<number>
Execute INSERT/UPDATE/DELETE and return affected row count.
executeBatch(
sql: string,
paramsArray: unknown[][]
): Promise<number[]>
Execute multiple commands in a batch. Implementation should optimize this if possible (e.g. using prepared statements or multi-row inserts).
query<T extends Record<string, any> = Record<string, any>>(
sql: string,
params?: unknown[]
): Promise<T[]>
Execute a SELECT query and return rows.
stream<T extends Record<string, any> = Record<string, any>>(
sql: string,
params?: unknown[]
): AsyncIterable<T>
Stream rows from a SELECT query using a cursor.

I

Connector that produces connected DBAdapter instances.

connect(signal?: AbortSignal): Promise<DBAdapter>
Acquire a connected adapter.

I

DBTransactionAdapter

Transaction adapter interface.

commit(): Promise<void>
Commit the transaction.
execute(
sql: string,
params?: unknown[]
): Promise<number>
Execute INSERT/UPDATE/DELETE and return affected row count.
executeBatch(
sql: string,
paramsArray: unknown[][]
): Promise<number[]>
Execute multiple commands in a batch.
query<T extends Record<string, any> = Record<string, any>>(
sql: string,
params?: unknown[]
): Promise<T[]>
Execute a SELECT query and return rows.
rollback(): Promise<void>
Rollback the transaction.

Type Aliases

T

Notification = { channel: string; payload: string; }

A notification received from a pub/sub channel.

channel: string
The channel the notification was sent to.
payload: string
The payload string.

Variables

v

DBErrors: { ConnectionFailed; QueryFailed; ConstraintViolation; TransactionFailed; CloseError; SerializationFailure; ListenFailed; }

Registry of error constructors for instanceof checks.