feat(core): add em.begin/commit/rollback methods

docs/docs/transactions.md

@@ -47,14 +47,33 @@ The explicit alternative is to use the transactions API directly to control the
 The code then looks like this:
 
 ```typescript
-await orm.em.transactional(_em => {
+await orm.em.transactional(em => {
   //... do some work
   const user = new User(...);
   user.name = 'George';
-  _em.persistLater(user);
+  em.persist(user);
 });
 ```
 
+Or you can use `begin/commit/rollback` methods explicitly. Following example is
+equivalent to the previous one:
+
+```typescript
+const em = orm.em.fork(false);
+await em.begin();
+
+try {
+  //... do some work
+  const user = new User(...);
+  user.name = 'George';
+  em.persist(user);
+  await em.commit(); // will flush before making the actual commit query
+} catch (e) {
+  await em.rollback();
+  throw e;
+}
+```
+
 Explicit transaction demarcation is required when you want to include custom DBAL operations 
 in a unit of work or when you want to make use of some methods of the EntityManager API 
 that require an active transaction. Such methods will throw a `ValidationError` to inform 

packages/core/src/EntityManager.ts

@@ -286,6 +286,30 @@ export class EntityManager<D extends IDatabaseDriver = IDatabaseDriver> {
     }, ctx);
   }
 
+  /**
+   * Starts new transaction bound to this EntityManager. Use `ctx` parameter to provide the parent when nesting transactions.
+   */
+  async begin(ctx?: Transaction): Promise<void> {
+    this.transactionContext = await this.getConnection('write').begin(ctx);
+  }
+
+  /**
+   * Commits the transaction bound to this EntityManager. Flushes before doing the actual commit query.
+   */
+  async commit(): Promise<void> {
+    await this.flush();
+    await this.getConnection('write').commit(this.transactionContext);
+    delete this.transactionContext;
+  }
+
+  /**
+   * Rollbacks the transaction bound to this EntityManager.
+   */
+  async rollback(): Promise<void> {
+    await this.getConnection('write').rollback(this.transactionContext);
+    delete this.transactionContext;
+  }
+
   /**
    * Runs your callback wrapped inside a database transaction.
    */

packages/core/src/connections/Connection.ts

@@ -44,6 +44,18 @@ export abstract class Connection {
     throw new Error(`Transactions are not supported by current driver`);
   }
 
+  async begin(ctx?: Transaction): Promise<unknown> {
+    throw new Error(`Transactions are not supported by current driver`);
+  }
+
+  async commit(ctx: Transaction): Promise<void> {
+    throw new Error(`Transactions are not supported by current driver`);
+  }
+
+  async rollback(ctx: Transaction): Promise<void> {
+    throw new Error(`Transactions are not supported by current driver`);
+  }
+
   abstract async execute(query: string, params?: any[], method?: 'all' | 'get' | 'run', ctx?: Transaction): Promise<QueryResult | any | any[]>;
 
   getConnectionOptions(): ConnectionConfig {

packages/knex/src/AbstractSqlConnection.ts

@@ -1,4 +1,4 @@
-import Knex, { Config, QueryBuilder, Raw } from 'knex';
+import Knex, { Config, QueryBuilder, Raw, Transaction as KnexTransaction } from 'knex';
 import { readFile } from 'fs-extra';
 
 import { Connection, EntityData, AnyEntity, QueryResult, Transaction, Utils } from '@mikro-orm/core';
@@ -28,6 +28,18 @@ export abstract class AbstractSqlConnection extends Connection {
     return (ctx || this.client).transaction(cb);
   }
 
+  async begin(ctx?: KnexTransaction): Promise<KnexTransaction> {
+    return (ctx || this.client).transaction();
+  }
+
+  async commit(ctx: KnexTransaction): Promise<void> {
+    return ctx.commit();
+  }
+
+  async rollback(ctx: KnexTransaction): Promise<void> {
+    return ctx.rollback();
+  }
+
   async execute<T extends QueryResult | EntityData<AnyEntity> | EntityData<AnyEntity>[] = EntityData<AnyEntity>[]>(queryOrKnex: string | QueryBuilder | Raw, params: any[] = [], method: 'all' | 'get' | 'run' = 'all', ctx?: Transaction): Promise<T> {
     if (Utils.isObject<QueryBuilder | Raw>(queryOrKnex)) {
       if (ctx) {

packages/mongodb/src/MongoConnection.ts

@@ -1,4 +1,7 @@
-import { Collection, Db, DeleteWriteOpResultObject, InsertOneWriteOpResult, MongoClient, MongoClientOptions, ObjectId, UpdateWriteOpResult, FilterQuery as MongoFilterQuery, ClientSession } from 'mongodb';
+import {
+  Collection, Db, DeleteWriteOpResultObject, InsertOneWriteOpResult, MongoClient, MongoClientOptions,
+  ObjectId, UpdateWriteOpResult, FilterQuery as MongoFilterQuery, ClientSession,
+} from 'mongodb';
 import { inspect } from 'util';
 import {
   Connection, ConnectionConfig, QueryResult, Transaction, Utils, QueryOrder, QueryOrderMap,
@@ -153,6 +156,24 @@ export class MongoConnection extends Connection {
     return ret;
   }
 
+  async begin(ctx?: ClientSession): Promise<ClientSession> {
+    const session = ctx || this.client.startSession();
+    session.startTransaction();
+    this.logQuery('db.begin();');
+
+    return session;
+  }
+
+  async commit(ctx: ClientSession): Promise<void> {
+    await ctx.commitTransaction();
+    this.logQuery('db.commit();');
+  }
+
+  async rollback(ctx: ClientSession): Promise<void> {
+    await ctx.abortTransaction();
+    this.logQuery('db.rollback();');
+  }
+
   protected logQuery(query: string, took?: number): void {
     super.logQuery(query, took, 'javascript');
   }

tests/Connection.test.ts

@@ -31,6 +31,9 @@ describe('Connection', () => {
   test('by default it throws when trying to use transactions', async () => {
     const conn = new CustomConnection(new Configuration({ type: 'mongo' }, false));
     await expect(conn.transactional(async () => void 0)).rejects.toThrowError('Transactions are not supported by current driver');
+    await expect(conn.begin()).rejects.toThrowError('Transactions are not supported by current driver');
+    await expect(conn.commit({} as any)).rejects.toThrowError('Transactions are not supported by current driver');
+    await expect(conn.rollback({} as any)).rejects.toThrowError('Transactions are not supported by current driver');
   });
 
 });

tests/EntityManager.mongo.test.ts

@@ -339,38 +339,38 @@ describe('EntityManagerMongo', () => {
 
   test('transactions', async () => {
     const god1 = new Author('God1', 'hello@heaven1.god');
-    try {
-      await orm.em.transactional(async em => {
-        await em.persistAndFlush(god1);
-        throw new Error(); // rollback the transaction
-      });
-    } catch { }
-
+    await orm.em.begin();
+    await orm.em.persist(god1);
+    await orm.em.rollback();
     const res1 = await orm.em.findOne(Author, { name: 'God1' });
     expect(res1).toBeNull();
 
-    const ret = await orm.em.transactional(async em => {
-      const god2 = new Author('God2', 'hello@heaven2.god');
-      await em.persist(god2);
-      return true;
-    });
-
+    await orm.em.begin();
+    const god2 = new Author('God2', 'hello@heaven2.god');
+    orm.em.persist(god2);
+    await orm.em.commit();
     const res2 = await orm.em.findOne(Author, { name: 'God2' });
     expect(res2).not.toBeNull();
-    expect(ret).toBe(true);
+
+    await orm.em.transactional(async em => {
+      const god3 = new Author('God3', 'hello@heaven3.god');
+      em.persist(god3);
+    });
+    const res3 = await orm.em.findOne(Author, { name: 'God3' });
+    expect(res3).not.toBeNull();
 
     const err = new Error('Test');
 
     try {
       await orm.em.transactional(async em => {
-        const god3 = new Author('God4', 'hello@heaven4.god');
-        await em.persist(god3);
+        const god4 = new Author('God4', 'hello@heaven4.god');
+        em.persist(god4);
         throw err;
       });
     } catch (e) {
       expect(e).toBe(err);
-      const res3 = await orm.em.findOne(Author, { name: 'God4' });
-      expect(res3).toBeNull();
+      const res4 = await orm.em.findOne(Author, { name: 'God4' });
+      expect(res4).toBeNull();
     }
   });
 

tests/EntityManager.mysql.test.ts

@@ -260,38 +260,38 @@ describe('EntityManagerMySql', () => {
 
   test('transactions', async () => {
     const god1 = new Author2('God1', 'hello@heaven1.god');
-    try {
-      await orm.em.transactional(async em => {
-        await em.persistAndFlush(god1);
-        throw new Error(); // rollback the transaction
-      });
-    } catch { }
-
+    await orm.em.begin();
+    await orm.em.persist(god1);
+    await orm.em.rollback();
     const res1 = await orm.em.findOne(Author2, { name: 'God1' });
     expect(res1).toBeNull();
 
-    const ret = await orm.em.transactional(async em => {
-      const god2 = new Author2('God2', 'hello@heaven2.god');
-      await em.persist(god2);
-      return true;
-    });
-
+    await orm.em.begin();
+    const god2 = new Author2('God2', 'hello@heaven2.god');
+    orm.em.persist(god2);
+    await orm.em.commit();
     const res2 = await orm.em.findOne(Author2, { name: 'God2' });
     expect(res2).not.toBeNull();
-    expect(ret).toBe(true);
+
+    await orm.em.transactional(async em => {
+      const god3 = new Author2('God3', 'hello@heaven3.god');
+      await em.persist(god3);
+    });
+    const res3 = await orm.em.findOne(Author2, { name: 'God3' });
+    expect(res3).not.toBeNull();
 
     const err = new Error('Test');
 
     try {
       await orm.em.transactional(async em => {
-        const god3 = new Author2('God4', 'hello@heaven4.god');
-        await em.persist(god3);
+        const god4 = new Author2('God4', 'hello@heaven4.god');
+        await em.persist(god4);
         throw err;
       });
     } catch (e) {
       expect(e).toBe(err);
-      const res3 = await orm.em.findOne(Author2, { name: 'God4' });
-      expect(res3).toBeNull();
+      const res4 = await orm.em.findOne(Author2, { name: 'God4' });
+      expect(res4).toBeNull();
     }
   });