28658248094

Committed 03 Jul 2026 11:41AM UTC coverage: 96.61% (-1.0%) from 97.611%

Build # 28658248094

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #79 from palcarazm/issue/60

refactor: replace environment mocks with Testcontainers for integration tests

Coverage Stats

41 of 47 branches covered (87.23%)

Branch coverage included in aggregate %.

0 of 2 new or added lines in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

244 of 248 relevant lines covered (98.39%)

40.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.7

/src/main/postgresql/classes/PostgresBatchEntityReader.ts

import { Pool, PoolClient } from "pg";
import { AbstractBatchEntityReaderStream, AbstractBatchEntityReaderStreamOptions } from "../../common/index.js";
import { BatchData } from "batchjs";

/**
 * @interface
 * Options for the PostgresBatchEntityReader.
 * @extends AbstractBatchEntityReaderStreamOptions
 * @template T The type of the data to be read
 * @template E The type of the row data from the database
 */
export interface PostgresBatchEntityReaderOptions<T,E> extends AbstractBatchEntityReaderStreamOptions {
    /** The PostgreSQL connection pool */
    pool: Pool;
    /** SQL query to be executed (without LIMIT and OFFSET) */
    query: string;
    /** Function that converts a row to an entity */
    rowToEntity:(row: E) => T
}

/**
 * @class
 * Class that reads data in batches of a specified size using PostgreSQL cursors.
 * @extends AbstractBatchEntityReaderStream
 * @template T The type of the data to be read
 * @template E The type of the row data from the database
 */
export class PostgresBatchEntityReader<T,E> extends AbstractBatchEntityReaderStream<T> {
    private readonly pool: Pool;
    private readonly query: string;
    private readonly rowToEntity:(row: E) => T;
    private client: PoolClient | null = null;
    private cursorName: string | null = null;
    private cursorOpened: boolean = false;

    /**
     
     * @param {PostgresBatchEntityReaderOptions} options - The options for the PostgresBatchEntityReader.
     */
    constructor(options: PostgresBatchEntityReaderOptions<T,E>) {
        super(options);
        this.pool = options.pool;
        this.query = options.query;
        this.rowToEntity = options.rowToEntity;
    }

    /**
     * Fetches a batch of data using a PostgreSQL cursor.
     * 
     * @protected
     * @param {number} size - The size of the batch to fetch.
     * @returns {Promise<BatchData<T>>} A promise that resolves with the batch of data.
     */
    protected fetch(size: number): Promise<BatchData<T>> {
        return this.initializeCursor()
            .then(({cursorName,client})=>client.query({
                text: `FETCH ${size} FROM ${cursorName};`,
            }))
            .then((result) => result.rows.map(this.rowToEntity));
    }

    /**
     * Initializes the cursor for the query if not already initialized.
     * 
     * @private
     * @returns {Promise<{cursorName:string,client:PoolClient}>} A promise that resolves when the cursor is initialized.
     */
    private async initializeCursor(): Promise<{cursorName:string,client:PoolClient}> {
        if (!this.cursorOpened ) {
            this.cursorName = `cursor_${Date.now()}`;
            return this.pool.connect()
                .then((client) => {
                    this.client = client;
                    return this.client.query("BEGIN;");
                })
                .then(() => (this.client as PoolClient).query(`DECLARE ${this.cursorName} CURSOR FOR ${this.query};`))
                .then(() => {
                    this.cursorOpened = true;
                })
                .then(() => ({cursorName:this.cursorName as string,client:this.client as PoolClient}));
        }
        return Promise.resolve({cursorName:this.cursorName as string,client:this.client as PoolClient});
    }

    /**
     * Closes the cursor and releases the client connection.
     * 
     * @private
     * @returns {Promise<void>} A promise that resolves when the cursor is closed.
     */
    private async closeCursor(): Promise<void> {
        if (this.cursorOpened && this.cursorName && this.client) {
            try {
                await this.client.query(`CLOSE ${this.cursorName};`);
                await this.client.query("COMMIT;");
            } catch (error) {
                await this.client.query("ROLLBACK;");
                throw error;
            } finally {
                this.client.release();
                this.client = null;
                this.cursorName = null;
                this.cursorOpened = false;
            }
        }
    }

    /**
     * Destroys the reader by closing the cursor and releasing resources.
     * 
     * @param {Error|null} error - The error that caused the destruction.
     * @param {Function} callback - The callback function to be executed after destruction.
     */
    _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
        let destroyError = error;

        this.closeCursor()
            .catch((error) => {destroyError = error;})
            .finally(() => super._destroy(destroyError, callback));
    }
}

1	import { Pool, PoolClient } from "pg";
2	import { AbstractBatchEntityReaderStream, AbstractBatchEntityReaderStreamOptions } from "../../common/index.js";	12✔
3	import { BatchData } from "batchjs";
4
5	/**
6	* @interface
7	* Options for the PostgresBatchEntityReader.
8	* @extends AbstractBatchEntityReaderStreamOptions
9	* @template T The type of the data to be read
10	* @template E The type of the row data from the database
11	*/
12	export interface PostgresBatchEntityReaderOptions<T,E> extends AbstractBatchEntityReaderStreamOptions {
13	/** The PostgreSQL connection pool */
14	pool: Pool;
15	/** SQL query to be executed (without LIMIT and OFFSET) */
16	query: string;
17	/** Function that converts a row to an entity */
18	rowToEntity:(row: E) => T
19	}
20
21	/**
22	* @class
23	* Class that reads data in batches of a specified size using PostgreSQL cursors.
24	* @extends AbstractBatchEntityReaderStream
25	* @template T The type of the data to be read
26	* @template E The type of the row data from the database
27	*/
28	export class PostgresBatchEntityReader<T,E> extends AbstractBatchEntityReaderStream<T> {	12✔
29	private readonly pool: Pool;
30	private readonly query: string;
31	private readonly rowToEntity:(row: E) => T;
32	private client: PoolClient \| null = null;	18✔
33	private cursorName: string \| null = null;	18✔
34	private cursorOpened: boolean = false;	18✔
35
36	/**
37
38	* @param {PostgresBatchEntityReaderOptions} options - The options for the PostgresBatchEntityReader.
39	*/
40	constructor(options: PostgresBatchEntityReaderOptions<T,E>) {
41	super(options);	18✔
42	this.pool = options.pool;	18✔
43	this.query = options.query;	18✔
44	this.rowToEntity = options.rowToEntity;	18✔
45	}
46
47	/**
48	* Fetches a batch of data using a PostgreSQL cursor.
49	*
50	* @protected
51	* @param {number} size - The size of the batch to fetch.
52	* @returns {Promise<BatchData<T>>} A promise that resolves with the batch of data.
53	*/
54	protected fetch(size: number): Promise<BatchData<T>> {
55	return this.initializeCursor()	30✔
56	.then(({cursorName,client})=>client.query({	30✔
57	text: `FETCH ${size} FROM ${cursorName};`,
58	}))
59	.then((result) => result.rows.map(this.rowToEntity));	30✔
60	}
61
62	/**
63	* Initializes the cursor for the query if not already initialized.
64	*
65	* @private
66	* @returns {Promise<{cursorName:string,client:PoolClient}>} A promise that resolves when the cursor is initialized.
67	*/
68	private async initializeCursor(): Promise<{cursorName:string,client:PoolClient}> {
69	if (!this.cursorOpened ) {	30✔
70	this.cursorName = `cursor_${Date.now()}`;	18✔
71	return this.pool.connect()	18✔
72	.then((client) => {
73	this.client = client;	18✔
74	return this.client.query("BEGIN;");	18✔
75	})
76	.then(() => (this.client as PoolClient).query(`DECLARE ${this.cursorName} CURSOR FOR ${this.query};`))	18✔
77	.then(() => {
78	this.cursorOpened = true;	18✔
79	})
80	.then(() => ({cursorName:this.cursorName as string,client:this.client as PoolClient}));	18✔
81	}
82	return Promise.resolve({cursorName:this.cursorName as string,client:this.client as PoolClient});	12✔
83	}
84
85	/**
86	* Closes the cursor and releases the client connection.
87	*
88	* @private
89	* @returns {Promise<void>} A promise that resolves when the cursor is closed.
90	*/
91	private async closeCursor(): Promise<void> {
92	if (this.cursorOpened && this.cursorName && this.client) {	18!
93	try {	18✔
94	await this.client.query(`CLOSE ${this.cursorName};`);	18✔
95	await this.client.query("COMMIT;");	18✔
96	} catch (error) {
NEW 97	await this.client.query("ROLLBACK;");	×
NEW 98	throw error;	×
99	} finally {
100	this.client.release();	18✔
101	this.client = null;	18✔
102	this.cursorName = null;	18✔
103	this.cursorOpened = false;	18✔
104	}
105	}
106	}
107
108	/**
109	* Destroys the reader by closing the cursor and releasing resources.
110	*
111	* @param {Error\|null} error - The error that caused the destruction.
112	* @param {Function} callback - The callback function to be executed after destruction.
113	*/
114	_destroy(error: Error \| null, callback: (error?: Error \| null) => void): void {
115	let destroyError = error;	18✔
116
117	this.closeCursor()	18✔
UNCOV 118	.catch((error) => {destroyError = error;})	×
119	.finally(() => super._destroy(destroyError, callback));	18✔
120	}
121	}

palcarazm / batchjs-data / 28658248094

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous