feat: implement mysql execute

src/client.ts

@@ -79,6 +79,7 @@ export class RDSClient extends Operator {
       'query',
       'getConnection',
       'end',
+      'execute',
     ].forEach(method => {
       this.#pool[method] = promisify(this.#pool[method]);
     });
@@ -113,30 +114,45 @@ export class RDSClient extends Operator {
   }
 
   async query<T = any>(sql: string, values?: object | any[], options?: QueryOptions): Promise<T> {
-    let conn: RDSConnection | RDSTransaction;
-    let shouldReleaseConn = false;
-    if (options?.conn) {
-      conn = options.conn;
-    } else {
-      const ctx = this.#connectionStorage.getStore();
-      const ctxConn = ctx?.[this.#connectionStorageKey];
-      if (ctxConn) {
-        conn = ctxConn;
-      } else {
-        conn = await this.getConnection();
-        shouldReleaseConn = true;
-      }
-    }
+    return this.#executeWithConnection('query', sql, values, options);
+  }
+
+  async execute<T = any>(sql: string, values?: object | any[], options?: QueryOptions): Promise<T> {
+    return this.#executeWithConnection('execute', sql, values, options);
+  }
+
+  async #executeWithConnection<T = any>(
+    method: 'query' | 'execute',
+    sql: string,
+    values?: object | any[],
+    options?: QueryOptions,
+  ): Promise<T> {
+    const { conn, shouldRelease } = await this.#getConnection(options);
 
     try {
-      return await conn.query(sql, values);
+      return await conn[method](sql, values);
     } finally {
-      if (shouldReleaseConn) {
+      if (shouldRelease) {
         (conn as RDSConnection).release();
       }
     }
   }
 
+  async #getConnection(options?: QueryOptions): Promise<{ conn: RDSConnection | RDSTransaction; shouldRelease: boolean }> {
+    if (options?.conn) {
+      return { conn: options.conn, shouldRelease: false };
+    }
+
+    const ctx = this.#connectionStorage.getStore();
+    const ctxConn = ctx?.[this.#connectionStorageKey];
+    if (ctxConn) {
+      return { conn: ctxConn, shouldRelease: false };
+    }
+
+    const conn = await this.getConnection();
+    return { conn, shouldRelease: true };
+  }
+
   get pool() {
     return this.#pool;
   }

src/connection.ts

@@ -16,6 +16,7 @@ export class RDSConnection extends Operator {
     if (!this.conn[kWrapToRDS]) {
       [
         'query',
+        'execute',
         'beginTransaction',
         'commit',
         'rollback',
@@ -36,6 +37,10 @@ export class RDSConnection extends Operator {
     return await this.conn.query(sql, values);
   }
 
+  async _execute(sql: string, values?: object | any[]) {
+    return await this.conn.execute(sql, values);
+  }
+
   async beginTransaction() {
     return await this.conn.beginTransaction();
   }

src/operator.ts

@@ -74,7 +74,30 @@ export abstract class Operator {
   }
 
   async query<T = any>(sql: string, values?: object | any[]): Promise<T> {
-    // query(sql, values)
+    return this.#executeWithHooks(sql, values, 'query', this._query.bind(this));
+  }
+
+  async queryOne(sql: string, values?: object | any[]) {
+    const rows = await this.query(sql, values);
+    return rows && rows[0] || null;
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  protected async _query(_sql: string, _values?: object | any[]): Promise<any> {
+    throw new Error('SubClass must impl this');
+  }
+
+  async execute<T = any>(sql: string, values?: object | any[]): Promise<T> {
+    return this.#executeWithHooks(sql, values, 'execute', this._execute.bind(this));
+  }
+
+  async #executeWithHooks<T = any>(
+    sql: string,
+    values: object | any[] | undefined,
+    operation: string,
+    executor: (sql: string, values?: object | any[]) => Promise<any>,
+  ): Promise<T> {
+    // 处理前置钩子
     if (this.beforeQueryHandlers.length > 0) {
       for (const beforeQueryHandler of this.beforeQueryHandlers) {
         const newSql = beforeQueryHandler(sql, values);
@@ -83,30 +106,35 @@ export abstract class Operator {
         }
       }
     }
-    debug('[connection#%s] query %o', this.threadId, sql);
+
+    debug('[connection#%s] %s %o', this.threadId, operation, sql);
     if (typeof this.logging === 'function') {
       this.logging(sql, { threadId: this.threadId });
     }
+
     const queryStart = performance.now();
     let rows: any;
     let lastError: Error | undefined;
+
     channels.queryStart.publish({
       sql,
       values,
       connection: this.#connection,
     } as QueryStartMessage);
+
     try {
-      rows = await this._query(sql, values);
+      rows = await executor(sql, values);
+
       if (Array.isArray(rows)) {
-        debug('[connection#%s] query get %o rows', this.threadId, rows.length);
+        debug('[connection#%s] %s get %o rows', this.threadId, operation, rows.length);
       } else {
-        debug('[connection#%s] query result: %o', this.threadId, rows);
+        debug('[connection#%s] %s result: %o', this.threadId, operation, rows);
       }
       return rows;
     } catch (err) {
       lastError = err;
       err.stack = `${err.stack}\n    sql: ${sql}`;
-      debug('[connection#%s] query error: %o', this.threadId, err);
+      debug('[connection#%s] %s error: %o', this.threadId, operation, err);
       throw err;
     } finally {
       const duration = Math.floor((performance.now() - queryStart) * 1000) / 1000;
@@ -117,6 +145,7 @@ export abstract class Operator {
         duration,
         error: lastError,
       } as QueryEndMessage);
+
       if (this.afterQueryHandlers.length > 0) {
         for (const afterQueryHandler of this.afterQueryHandlers) {
           afterQueryHandler(sql, rows, duration, lastError, values);
@@ -125,13 +154,8 @@ export abstract class Operator {
     }
   }
 
-  async queryOne(sql: string, values?: object | any[]) {
-    const rows = await this.query(sql, values);
-    return rows && rows[0] || null;
-  }
-
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  protected async _query(_sql: string, _values?: object | any[]): Promise<any> {
+  protected async _execute(_sql: string, _values?: object | any[]): Promise<any> {
     throw new Error('SubClass must impl this');
   }
 

src/transaction.ts

@@ -41,6 +41,11 @@ export class RDSTransaction extends Operator {
     return await this.conn!._query(sql, values);
   }
 
+  async _execute(sql: string, values?: object | any[]) {
+    this.#check();
+    return await this.conn!._execute(sql, values);
+  }
+
   #check() {
     if (!this.conn) {
       throw new Error('transaction was commit or rollback');

src/types.ts

@@ -12,8 +12,9 @@ export interface RDSClientOptions extends PoolOptions {
   logging?: Logging;
 }
 
-export interface PoolConnectionPromisify extends Omit<PoolConnection, 'query'> {
+export interface PoolConnectionPromisify extends Omit<PoolConnection, 'query' | 'execute'> {
   query(sql: string, values?: any | any[] | { [param: string]: any }): Promise<any>;
+  execute(sql: string, values?: any | any[] | { [param: string]: any }): Promise<any>;
   beginTransaction(): Promise<void>;
   commit(): Promise<void>;
   rollback(): Promise<void>;

test/client.test.ts

@@ -1617,4 +1617,96 @@ describe('test/client.test.ts', () => {
     });
 
   });
+
+  describe('execute()', () => {
+    it('should execute sql with parameters', async () => {
+      const result = await db.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+        [ prefix + 'execute-test', prefix + 'm@execute-test.com' ]);
+      assert.equal(result.affectedRows, 1);
+      assert(result.insertId > 0);
+    });
+
+    it('should execute select query', async () => {
+      await db.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+        [ prefix + 'execute-select-test', prefix + 'm@execute-select-test.com' ]);
+
+      const rows = await db.execute('SELECT * FROM `myrds-test-user` WHERE email = ?',
+        [ prefix + 'm@execute-select-test.com' ]);
+      assert(Array.isArray(rows));
+      assert.equal(rows.length, 1);
+      assert.equal(rows[0].name, prefix + 'execute-select-test');
+    });
+
+    it('should execute in transaction', async () => {
+      const tran = await db.beginTransaction();
+      try {
+        const result = await db.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+          [ prefix + 'execute-transaction-test', prefix + 'm@execute-transaction-test.com' ],
+          { conn: tran });
+        assert.equal(result.affectedRows, 1);
+        await tran.commit();
+      } catch (err) {
+        await tran.rollback();
+        throw err;
+      }
+
+      const rows = await db.execute('SELECT * FROM `myrds-test-user` WHERE email = ?',
+        [ prefix + 'm@execute-transaction-test.com' ]);
+      assert.equal(rows.length, 1);
+    });
+
+    it('should execute in transaction scope', async () => {
+      await db.beginTransactionScope(async tran => {
+        const result = await db.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+          [ prefix + 'execute-scope-test', prefix + 'm@execute-scope-test.com' ],
+          { conn: tran });
+        assert.equal(result.affectedRows, 1);
+      });
+
+      const rows = await db.execute('SELECT * FROM `myrds-test-user` WHERE email = ?',
+        [ prefix + 'm@execute-scope-test.com' ]);
+      assert.equal(rows.length, 1);
+    });
+
+    it('should execute with connection', async () => {
+      const conn = await db.getConnection();
+      try {
+        const result = await conn.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+          [ prefix + 'execute-conn-test', prefix + 'm@execute-conn-test.com' ]);
+        assert.equal(result.affectedRows, 1);
+      } finally {
+        conn.release();
+      }
+
+      const rows = await db.execute('SELECT * FROM `myrds-test-user` WHERE email = ?',
+        [ prefix + 'm@execute-conn-test.com' ]);
+      assert.equal(rows.length, 1);
+    });
+
+    it('should execute update query', async () => {
+      await db.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+        [ prefix + 'execute-update-test', prefix + 'm@execute-update-test.com' ]);
+
+      const result = await db.execute('UPDATE `myrds-test-user` SET email = ? WHERE name = ?',
+        [ prefix + 'm@execute-updated.com', prefix + 'execute-update-test' ]);
+      assert.equal(result.affectedRows, 1);
+
+      const rows = await db.execute('SELECT * FROM `myrds-test-user` WHERE email = ?',
+        [ prefix + 'm@execute-updated.com' ]);
+      assert.equal(rows.length, 1);
+    });
+
+    it('should execute delete query', async () => {
+      await db.execute('INSERT INTO `myrds-test-user` (name, email, gmt_create, gmt_modified) VALUES(?, ?, now(), now())',
+        [ prefix + 'execute-delete-test', prefix + 'm@execute-delete-test.com' ]);
+
+      const result = await db.execute('DELETE FROM `myrds-test-user` WHERE name = ?',
+        [ prefix + 'execute-delete-test' ]);
+      assert.equal(result.affectedRows, 1);
+
+      const rows = await db.execute('SELECT * FROM `myrds-test-user` WHERE name = ?',
+        [ prefix + 'execute-delete-test' ]);
+      assert.equal(rows.length, 0);
+    });
+  });
 });