Function stream

packages/sql/src/AbstractSqlConnection.ts:311–353

Executes a SQL query and returns an async iterable that yields results row by row.

(
    query: string | NativeQueryBuilder | RawQueryFragment,
    params: readonly unknown[] = [],
    ctx?: Transaction<Kysely<any>>,
    loggerContext?: LoggingOptions,
    chunkSize?: number,
  )

  /** Executes a SQL query and returns an async iterable that yields results row by row. */
  async *stream<T extends EntityData<AnyEntity>>(
    query: string | NativeQueryBuilder | RawQueryFragment,
    params: readonly unknown[] = [],
    ctx?: Transaction<Kysely<any>>,
    loggerContext?: LoggingOptions,
    chunkSize?: number,
  ): AsyncIterableIterator<T> {
    await this.ensureConnection();
    const q = this.prepareQuery(query, params);
    const sql = this.getSql(q.query, q.formatted, loggerContext);
    const { abort, loggerContext: cleanCtx } = extractAbortOptions(loggerContext);

    // construct the compiled query manually with `kind: 'SelectQueryNode'` to avoid sqlite validation for select queries when streaming
    const compiled = {
      query: {
        kind: 'SelectQueryNode',
      },
      sql: q.formatted,
      parameters: [],
    } as unknown as CompiledQuery;

    try {
      const res = (ctx ?? this.getClient())
        .getExecutor()
        .stream(compiled, chunkSize ?? 100, abort ? { signal: abort.signal } : undefined);

      this.logQuery(sql, {
        sql,
        params,
        ...cleanCtx,
        affected: Utils.isPlainObject<QueryResult>(res) ? res.affectedRows : undefined,
      });

      for await (const items of res) {
        for (const row of this.transformRawResult(items, 'all') as T[]) {
          yield row;
        }
      }
    } catch (e) {
      this.logQuery(sql, { sql, params, ...cleanCtx, level: 'error' });
      throw e;
    }
  }

  /** @inheritDoc */
  override async executeDump(dump: string): Promise<void> {