Introduction to Multi-Document ACID Transactions in Python

1
$ cd pymongo-transactions
2
$ virtualenv -p python3 venv
3
$ source venv/bin/activate

1
pip install --upgrade pymongo

1
pip install mtools

1
pip install psutil

1
mlaunch init --port 27100 --replicaset --name "txntest"

1
$ cd pymongo-transactions
2
$ make

1
$ python transaction_main.py -h
2
usage: transaction_main.py [-h] [--host HOST] [--usetxns] [--delay DELAY]
3
                           [--iterations ITERATIONS]
4
                           [--randdelay RANDDELAY RANDDELAY]
5
6
optional arguments:
7
  -h, --help            show this help message and exit
8
  --host HOST           MongoDB URI [default: mongodb://localhost:27100,localh
9
                        ost:27101,localhost:27102/?replicaSet=txntest&retryWri
10
                        tes=true]
11
  --usetxns             Use transactions [default: False]
12
  --delay DELAY         Delay between two insertion events [default: 1.0]
13
  --iterations ITERATIONS
14
                        Run N iterations. O means run forever
15
  --randdelay RANDDELAY RANDDELAY
16
                        Create a delay set randomly between the two bounds
17
                        [default: None]

1
$ python transaction_main.py
2
using collection: SEATSDB.seats
3
using collection: PAYMENTSDB.payments
4
using collection: AUDITDB.audit
5
Using a fixed delay of 1.0
6
7
1. Booking seat: '1A'
8
1. Sleeping: 1.000
9
1. Paying 330 for seat '1A'
10
2. Booking seat: '2A'
11
2. Sleeping: 1.000
12
2. Paying 450 for seat '2A'
13
3. Booking seat: '3A'
14
3. Sleeping: 1.000
15
3. Paying 490 for seat '3A'
16
4. Booking seat: '4A'
17
4. Sleeping: 1.000

1
Traceback (most recent call last):
2
  File "transaction_main.py", line 175, in
3
    total_delay = total_delay + run_transaction_with_retry( booking_functor, session)
4
  File "/Users/jdrumgoole/GIT/pymongo-transactions/transaction_retry.py", line 52, in run_transaction_with_retry
5
    with session.start_transaction():
6
AttributeError: 'ClientSession' object has no attribute 'start_transaction'

1
$ python watch_transactions.py -h
2
usage: watch_transactions.py [-h] [--host HOST] [--collection COLLECTION]
3
4
optional arguments:
5
  -h, --help            show this help message and exit
6
  --host HOST           mongodb URI for connecting to server [default:
7
                        mongodb://localhost:27100/?replicaSet=txntest]
8
  --collection COLLECTION
9
                        Watch  [default:
10
                        PYTHON_TXNS_EXAMPLE.seats_collection]

1
$ python watch_transactions.py --watch seats
2
Watching: seats
3
...

1
$ python watch_transactions.py --watch payments
2
Watching: payments
3
...

1
def book_seat(seats, payments, audit, seat_no, delay_range, session=None):
2
    '''
3
    Run two inserts in sequence.
4
    If session is not None we are in a transaction
5
6
    :param seats: seats collection
7
    :param payments: payments collection
8
    :param seat_no: the number of the seat to be booked (defaults to row A)
9
    :param delay_range: A tuple indicating a random delay between two ranges or a single float fixed delay
10
    :param session: Session object required by a MongoDB transaction
11
    :return: the delay_period for this transaction
12
    '''
13
    price = random.randrange(200, 500, 10)
14
    if type(delay_range) == tuple:
15
        delay_period = random.uniform(delay_range[0], delay_range[1])
16
    else:
17
        delay_period = delay_range
18
19
    # Book Seat
20
    seat_str = "{}A".format(seat_no)
21
    print(count( i, "Booking seat: '{}'".format(seat_str)))
22
    seats.insert_one({"flight_no" : "EI178",
23
                      "seat"      : seat_str,
24
                      "date"      : datetime.datetime.utcnow()},
25
                     session=session)
26
    print(count( seat_no, "Sleeping: {:02.3f}".format(delay_period)))
27
    #pay for seat
28
    time.sleep(delay_period)
29
    payments.insert_one({"flight_no" : "EI178",
30
                         "seat"      : seat_str,
31
                         "date"      : datetime.datetime.utcnow(),
32
                         "price"     : price},
33
                        session=session)
34
    audit.update_one({ "audit" : "seats"}, { "$inc" : { "count" : 1}}, upsert=True)
35
    print(count(seat_no, "Paying {} for seat '{}'".format(price, seat_str)))
36
37
    return delay_period

1
$ python transaction_main.py

1
$ python transaction_main.py --usetxns

1
def commit_with_retry(session):
2
    while True:
3
        try:
4
            # Commit uses write concern set at transaction start.
5
            session.commit_transaction()
6
            print("Transaction committed.")
7
            break
8
        except (pymongo.errors.ConnectionFailure, pymongo.errors.OperationFailure) as exc:
9
            # Can retry commit
10
            if exc.has_error_label("UnknownTransactionCommitResult"):
11
                print("UnknownTransactionCommitResult, retrying "
12
                      "commit operation ...")
13
                continue
14
            else:
15
                print("Error during commit ...")
16
                raise
17
18
def run_transaction_with_retry(functor, session):
19
    assert (isinstance(functor, Transaction_Functor))
20
    while True:
21
        try:
22
            with session.start_transaction():
23
                result=functor(session)  # performs transaction
24
                commit_with_retry(session)
25
            break
26
        except (pymongo.errors.ConnectionFailure, pymongo.errors.OperationFailure) as exc:
27
            # If transient error, retry the whole transaction
28
            if exc.has_error_label("TransientTransactionError"):
29
                print("TransientTransactionError, retrying "
30
                      "transaction ...")
31
                continue
32
            else:
33
                raise
34
35
    return result

1
$ make kill_primary
2
. venv/bin/activate && python kill_primary.py
3
no nodes started.
4
Current electionTimeoutMillis: 500
5
1. (Re)starting replica-set
6
no nodes started.
7
1. Getting list of mongod processes
8
Process list written to mlaunch.procs
9
1. Getting replica set status
10
1. Killing primary node: 31029
11
1. Sleeping: 1.0
12
2. (Re)starting replica-set
13
launching: "/usr/local/mongodb/bin/mongod" on port 27101
14
2. Getting list of mongod processes
15
Process list written to mlaunch.procs
16
2. Getting replica set status
17
2. Killing primary node: 31045
18
2. Sleeping: 1.0
19
3. (Re)starting replica-set
20
launching: "/usr/local/mongodb/bin/mongod" on port 27102
21
3. Getting list of mongod processes
22
Process list written to mlaunch.procs
23
3. Getting replica set status
24
3. Killing primary node: 31137
25
3. Sleeping: 1.0

1
$ make usetxns
2
. venv/bin/activate && python transaction_main.py --usetxns
3
Forcing collection creation (you can't create collections inside a txn)
4
Collections created
5
using collection: PYTHON_TXNS_EXAMPLE.seats
6
using collection: PYTHON_TXNS_EXAMPLE.payments
7
using collection: PYTHON_TXNS_EXAMPLE.audit
8
Using a fixed delay of 1.0
9
Using transactions
10
11
1. Booking seat: '1A'
12
1. Sleeping: 1.000
13
1. Paying 440 for seat '1A'
14
Transaction committed.
15
2. Booking seat: '2A'
16
2. Sleeping: 1.000
17
2. Paying 330 for seat '2A'
18
Transaction committed.
19
3. Booking seat: '3A'
20
3. Sleeping: 1.000
21
TransientTransactionError, retrying transaction ...
22
3. Booking seat: '3A'
23
3. Sleeping: 1.000
24
3. Paying 240 for seat '3A'
25
Transaction committed.
26
4. Booking seat: '4A'
27
4. Sleeping: 1.000
28
4. Paying 410 for seat '4A'
29
Transaction committed.
30
5. Booking seat: '5A'
31
5. Sleeping: 1.000
32
5. Paying 260 for seat '5A'
33
Transaction committed.
34
6. Booking seat: '6A'
35
6. Sleeping: 1.000
36
TransientTransactionError, retrying transaction ...
37
6. Booking seat: '6A'
38
6. Sleeping: 1.000
39
6. Paying 380 for seat '6A'
40
Transaction committed.
41
...