sql-transaction-test.py

#!/usr/bin/python
#
# Create insert statements and do rollbacks to a database (bsc#1120593)
# TODO:
# - Create the database 'testdb'
# - Grant user access with something like: GRANT ALL PRIVILEGES ON testdb.* TO 'keystone'@'%';
#
# Author: Thomas Bechtold <[email protected]>
#

from multiprocessing import Process

from sqlalchemy import create_engine
from sqlalchemy import Table, MetaData, Column
from sqlalchemy import String, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.pool import Pool


BASE = declarative_base()

DB_NAME='testdb'
CONNECTION_STRING='mysql+pymysql://keystone:[email protected]/{}?charset=utf8'.format(DB_NAME)

class Data(BASE):
    __tablename__ = 'data'
    id = Column(Integer, primary_key=True)
    data = Column(String(255), index=True)


def _do_transactional_insert(engine):
    sm = sessionmaker(bind=engine)
    s = sm()
    d = Data(data="X"*255)
    s.add(d)
    s.commit()


def _do_transactional_failing_insert(engine, data_num):
    sm = sessionmaker(bind=engine)
    s = sm()
    for x in range(data_num):
        d_ok = Data(data="X"*200)
        s.add(d_ok)
    d_fail = Data(data="X"*500)
    s.add(d_fail)
    try:
        print("commit with {}".format(data_num))
        s.commit()
    except:
        print("rollback begin")
        s.rollback()
        print("rollback end")


def _do_endless(engine, data_num):
    while True:
        _do_transactional_failing_insert(engine, data_num)


if __name__ == "__main__":
    engine = create_engine(CONNECTION_STRING)
    session = sessionmaker()
    session.configure(bind=engine)
    BASE.metadata.create_all(engine, checkfirst=True)

    procs = []
    for pp in range(1, 4):
        p = Process(target=_do_endless, args=(engine, 100000,))
        procs.append(p)
        p.start()

    procs[0].join()