mirror of
https://github.com/shishantbiswas/bknd.git
synced 2026-03-17 12:56:05 +00:00
added batching for postgres
This commit is contained in:
@@ -21,7 +21,7 @@ describe.skipIf(ALL_TESTS)("postgres", () => {
|
|||||||
plugins: [new ParseJSONResultsPlugin()],
|
plugins: [new ParseJSONResultsPlugin()],
|
||||||
});
|
});
|
||||||
|
|
||||||
console.log(await introspector.getSchema());
|
console.log(await introspector.getSchemaSpec());
|
||||||
});
|
});
|
||||||
|
|
||||||
test("builds", async () => {
|
test("builds", async () => {
|
||||||
@@ -50,6 +50,11 @@ describe.skipIf(ALL_TESTS)("postgres", () => {
|
|||||||
|
|
||||||
await app.build({ sync: true });
|
await app.build({ sync: true });
|
||||||
|
|
||||||
|
/*await app.em
|
||||||
|
.mutator("posts")
|
||||||
|
.insertMany([{ title: "hello world" }, { title: "hello world 2" }]);*/
|
||||||
|
|
||||||
expect(app.version()).toBeDefined();
|
expect(app.version()).toBeDefined();
|
||||||
|
console.log(await app.em.repo("posts").findMany());
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -3,3 +3,11 @@ export function clampNumber(value: number, min: number, max: number): number {
|
|||||||
const upper = Math.max(min, max);
|
const upper = Math.max(min, max);
|
||||||
return Math.max(lower, Math.min(value, upper));
|
return Math.max(lower, Math.min(value, upper));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function ensureInt(value?: string | number | null | undefined): number {
|
||||||
|
if (value === undefined || value === null) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
return typeof value === "number" ? value : Number.parseInt(value, 10);
|
||||||
|
}
|
||||||
|
|||||||
@@ -151,4 +151,5 @@ export abstract class Connection<DB = any> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
abstract getFieldSchema(spec: FieldSpec, strict?: boolean): SchemaResponse;
|
abstract getFieldSchema(spec: FieldSpec, strict?: boolean): SchemaResponse;
|
||||||
|
abstract close(): Promise<void>;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,7 +8,12 @@ import {
|
|||||||
} from "kysely";
|
} from "kysely";
|
||||||
import pg from "pg";
|
import pg from "pg";
|
||||||
import { PostgresIntrospector } from "./PostgresIntrospector";
|
import { PostgresIntrospector } from "./PostgresIntrospector";
|
||||||
import { type FieldSpec, type SchemaResponse, Connection } from "data/connection/Connection";
|
import {
|
||||||
|
type FieldSpec,
|
||||||
|
type SchemaResponse,
|
||||||
|
Connection,
|
||||||
|
type QB,
|
||||||
|
} from "data/connection/Connection";
|
||||||
|
|
||||||
export type PostgresConnectionConfig = pg.PoolConfig;
|
export type PostgresConnectionConfig = pg.PoolConfig;
|
||||||
|
|
||||||
@@ -21,16 +26,20 @@ class CustomPostgresDialect extends PostgresDialect {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export class PostgresConnection extends Connection {
|
export class PostgresConnection extends Connection {
|
||||||
|
private pool: pg.Pool;
|
||||||
|
|
||||||
constructor(config: PostgresConnectionConfig) {
|
constructor(config: PostgresConnectionConfig) {
|
||||||
|
const pool = new pg.Pool(config);
|
||||||
const kysely = new Kysely({
|
const kysely = new Kysely({
|
||||||
dialect: new CustomPostgresDialect({
|
dialect: new CustomPostgresDialect({
|
||||||
pool: new pg.Pool(config),
|
pool,
|
||||||
}),
|
}),
|
||||||
plugins,
|
plugins,
|
||||||
//log: ["query", "error"],
|
//log: ["query", "error"],
|
||||||
});
|
});
|
||||||
|
|
||||||
super(kysely, {}, plugins);
|
super(kysely, {}, plugins);
|
||||||
|
this.pool = pool;
|
||||||
}
|
}
|
||||||
|
|
||||||
override supportsIndices(): boolean {
|
override supportsIndices(): boolean {
|
||||||
@@ -73,4 +82,22 @@ export class PostgresConnection extends Connection {
|
|||||||
},
|
},
|
||||||
];
|
];
|
||||||
}
|
}
|
||||||
|
|
||||||
|
override supportsBatching(): boolean {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
override async close(): Promise<void> {
|
||||||
|
await this.pool.end();
|
||||||
|
}
|
||||||
|
|
||||||
|
protected override async batch<Queries extends QB[]>(
|
||||||
|
queries: [...Queries],
|
||||||
|
): Promise<{
|
||||||
|
[K in keyof Queries]: Awaited<ReturnType<Queries[K]["execute"]>>;
|
||||||
|
}> {
|
||||||
|
return this.kysely.transaction().execute(async (trx) => {
|
||||||
|
return Promise.all(queries.map((q) => trx.executeQuery(q).then((r) => r.rows)));
|
||||||
|
}) as any;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -47,4 +47,8 @@ export class SqliteConnection extends Connection {
|
|||||||
},
|
},
|
||||||
] as const;
|
] as const;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
override async close(): Promise<void> {
|
||||||
|
// no-op
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,6 +14,7 @@ import {
|
|||||||
WithBuilder,
|
WithBuilder,
|
||||||
} from "../index";
|
} from "../index";
|
||||||
import { JoinBuilder } from "./JoinBuilder";
|
import { JoinBuilder } from "./JoinBuilder";
|
||||||
|
import { ensureInt } from "core/utils";
|
||||||
|
|
||||||
export type RepositoryQB = SelectQueryBuilder<any, any, any>;
|
export type RepositoryQB = SelectQueryBuilder<any, any, any>;
|
||||||
|
|
||||||
@@ -225,8 +226,9 @@ export class Repository<TBD extends object = DefaultDB, TB extends keyof TBD = a
|
|||||||
data,
|
data,
|
||||||
meta: {
|
meta: {
|
||||||
...payload.meta,
|
...payload.meta,
|
||||||
total: _total[0]?.total ?? 0,
|
// parsing is important since pg returns string
|
||||||
count: _count[0]?.count ?? 0, // @todo: better graceful method
|
total: ensureInt(_total[0]?.total),
|
||||||
|
count: ensureInt(_count[0]?.count),
|
||||||
items: result.length,
|
items: result.length,
|
||||||
time,
|
time,
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -329,6 +329,7 @@ export class SchemaManager {
|
|||||||
if (local_updates === 0) continue;
|
if (local_updates === 0) continue;
|
||||||
|
|
||||||
// iterate through built qbs
|
// iterate through built qbs
|
||||||
|
// @todo: run in batches
|
||||||
for (const qb of qbs) {
|
for (const qb of qbs) {
|
||||||
const { sql, parameters } = qb.compile();
|
const { sql, parameters } = qb.compile();
|
||||||
statements.push({ sql, parameters });
|
statements.push({ sql, parameters });
|
||||||
|
|||||||
Reference in New Issue
Block a user