@flowblade/source-kysely

A source adapter for Kysely.

Install

pnpm add @flowblade/source-kysely kysely

# Install optional drivers
# 01. for Ms SqlServer or Azure Sql Edge
pnpm add tedious tarn

Kysely supports

Quick start

import { KyselyDatasource } from "@flowblade/source-kysely";

// Your Kysely instance, see the Utils section for Ms SqlServer
import { db } from "@/config/db.config.ts";

const ds = new KyselyDatasource({ connection: db });
const query = ds.queryBuilder // This gives access to Kysely expression builder
  .selectFrom("brand as b")
  .select(["b.id", "b.name"])
  .leftJoin("product as p", "p.brand_id", "b.id")
  .select(["p.id as product_id", "p.name as product_name"])
  .where("b.created_at", "<", new Date())
  .orderBy("b.name", "desc");

const result = await ds.query(query);

// Or with query information (will be sent in the metadata)
// const result = await ds.query(query, {
//  name: 'getBrands'
// });

// Option 1: the QResult object contains the data, metadata and error
//  - data:  the result rows (TData or undefined if error)
//  - error: the error (QError or undefined if success)
//  - meta:  the metadata (always present)

const { data, meta, error } = result;

// Option 2: You operate over the result, ie: mapping the data

const { data: data2 } = result.map((row) => {
  return {
    ...row,
    key: `key-${row.product_id}`,
  };
});

// Or get the rows, throws on error (the driver error as cause)
const { data: rows } = await ds.queryOrThrow(query);

Streaming

import { KyselyDatasource } from "@flowblade/source-kysely";

const ds = new KyselyDatasource({ connection: db });
const query = ds.queryBuilder // This gives access to Kysely expression builder
  .selectFrom("brand as b")
  .select(["b.id", "b.name"]);

const stream = ds.stream(query, {
  name: "getBrands",
  // Chunksize used when reading the database (only some dialects, ie: postgres)
  // @default undefined
  chunkSize: undefined,
  // Optional abort signal, see "Aborting queries"
  signal: AbortSignal.timeout(30_000),
});

for await (const brand of stream) {
  console.log(brand.name);
  if (brand.name === "Something") {
    // Breaking or returning before the stream has ended will release
    // the database connection and invalidate the stream.
    break;
  }
}

A failed stream throws an Error with the driver error as cause.

Error handling

query() never throws, failures are returned in result.error with the original driver error as cause:

const result = await ds.query(query);

if (result.error) {
  console.log(result.error.message);
  // The original driver error, ie: a tedious RequestError for Ms SqlServer
  const cause = result.error.cause as { number?: number } | undefined;
  if (cause?.number === 1205) {
    // SQL Server deadlock, the query can be retried
  }
}

// The cause is never serialized, safe to send to clients
const json = result.toJsonifiable();

queryOrThrow() throws an Error with the driver error as cause.

Aborting queries

Both query() and stream() accept an AbortSignal:

const result = await ds.query(query, {
  signal: AbortSignal.timeout(5_000),
});
// An aborted query returns an error result, an aborted stream throws

By default the query keeps running on the database once aborted, only the result is ignored. Use inflightQueryAbortStrategy to cancel it on the database (ie: Ms SqlServer, Postgres, MySQL):

const ds = new KyselyDatasource({
  connection: db,
  // 'ignore query' (default) | 'cancel query' | 'kill session'
  inflightQueryAbortStrategy: "cancel query",
});

// Or per query
await ds.query(query, {
  signal,
  inflightQueryAbortStrategy: "cancel query",
});

Note: Dialects that can't cancel fall back to 'ignore query' with a warning. Postgres and MySQL cancel through an extra connection, Ms SqlServer through the same one.

Logging

Queries are logged with LogTape under the ["flowblade", "kysely"] category: debug when starting, info on success (time and affected rows), error on failure. Nothing is logged until LogTape is configured by your application:

import { flowbladeLogtapeKyselyConfig } from "@flowblade/source-kysely";
import { configure, getConsoleSink } from "@logtape/logtape";

await configure({
  sinks: { console: getConsoleSink() },
  loggers: [
    {
      category: flowbladeLogtapeKyselyConfig.categories,
      lowestLevel: "info",
      sinks: ["console"],
    },
  ],
});

Logs include the sql and its params. When params might contain sensitive data, leave them out of the logs (they stay available in the result meta):

const ds = new KyselyDatasource({ connection: db, logParams: false });

A custom LogTape logger can also be given with the logger option.

Utils

createKyselyMssqlDialect

Create a Kysely dialect for Ms SqlServer or Azure Sql Edge.

import {
  TediousConnUtils,
  createKyselyMssqlDialect,
} from "@flowblade/source-kysely";
import * as Tedious from "tedious";

const jdbcDsn =
  "sqlserver://localhost:1433;database=db;user=sa;password=pwd;trustServerCertificate=true;encrypt=false";
const tediousConfig = TediousConnUtils.fromJdbcDsn(jdbcDsn);

const dialect = createKyselyMssqlDialect({
  tediousConfig,
  // 👉 Optional tarn pool options
  poolOptions: {
    min: 0, // 👉 Minimum number of connections, default 0
    max: 10, // 👉 Maximum number of connections, default 10
    propagateCreateError: true, // 👉 Propagate connection creation errors, default false
  },
  dialectConfig: {
    /**
     * When true, connections are validated before being acquired from the pool,
     * resulting in additional requests to the database.
     *
     * In safe scenarios, this can be set to false to improve performance.
     *
     * Defaults to `true`.
     */
    validateConnections: true,
    /**
     * When true, connections are reset to their initial states when released back to the pool,
     * resulting in additional requests to the database.
     *
     * Defaults to `false`.
     */
    resetConnectionsOnRelease: false,
    // 👉 Override Tedious types to enhance compatibility and modern support
    tediousTypes: {
      ...Tedious.TYPES,
      // see https://github.com/kysely-org/kysely/issues/1161#issuecomment-2384539764
      NVarChar: Tedious.TYPES.VarChar,
      // see https://github.com/kysely-org/kysely/issues/1596#issuecomment-3341591075
      DateTime: Tedious.TYPES.DateTime2,
    },
  },
});

const db = new Kysely<DB>({
  dialect,
});

Note: For performance you can avoid connection roundtrips by setting validateConnections to false and resetConnectionOnRelease to false.

TediousConnUtils

fromJdbcDsn

Parse and validate a JDBC connection string and return a Tedious connection configuration.

import { TediousConnUtils } from "@flowblade/source-kysely";
import * as Tedious from "tedious";

// In your .env file
// DB_JDBC_DSN="sqlserver://localhost:1433;database=db;user=sa;password=pwd;trustServerCertificate=true;encrypt=false";

const tediousConfig = TediousConnUtils.fromJdbcDsn(process.env.DB_JDBC_DSN);

const tediousConnection = new Tedious.Connection(tediousConfig);

Compatibility

| Level | CI | Description | | --- | --- | --- | | Node | ✅ | Requires 22.18+, CI for 22.x, 24.x & 26.x | | Bun | ✅ | Bun 1.4+, unit tests run on CI with the latest Bun | | Deno | ✅ | Deno LTS, unit tests run on CI with Deno 2 | | Browserslist | ✅ | > 95% on 01/2025. Chrome 96+, Firefox 90+, Edge 19+, ios 15+, Safari 15+ and Opera 77+ | | Typescript | ✅ | TS 5.0 + / are-the-type-wrong checks on CI. | | ES2022 | ✅ | Dist files checked with es-check |

Contributors

Contributions are welcome. Have a look to the CONTRIBUTING document.

Sponsors

Sponsor, coffee, or star – All is spent for quality time with loved ones. Thanks ! 🙏❤️

Special thanks to

License

MIT © Sébastien Vanvelthem and contributors.