sqlalchemy-history/benchmark.py at master · bikmetle/sqlalchemy-history · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import itertools as it
import warnings
from copy import copy
from time import time

import sqlalchemy as sa
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, close_all_sessions
from sqlalchemy_history import make_versioned, versioning_manager, remove_versioning
from sqlalchemy_history.transaction import TransactionFactory
from sqlalchemy_history.plugins import (
    PropertyModTrackerPlugin,
    TransactionMetaPlugin,
    TransactionChangesPlugin,
)
from termcolor import colored

warnings.simplefilter("error", sa.exc.SAWarning)


def test_versioning(versioning_strategy, property_mod_tracking):
    transaction_column_name = "transaction_id"
    end_transaction_column_name = "end_transaction_id"
    plugins = [TransactionChangesPlugin(), TransactionMetaPlugin()]

    if property_mod_tracking:
        plugins.append(PropertyModTrackerPlugin())
    transaction_cls = TransactionFactory()
    user_cls = None

    Model = declarative_base()

    options = {
        "create_models": True,
        "base_classes": (Model,),
        "strategy": versioning_strategy,
        "transaction_column_name": transaction_column_name,
        "end_transaction_column_name": end_transaction_column_name,
    }

    make_versioned(options=options)

    dns = "postgresql://postgres:postgres@localhost/sqlalchemy_history_test"
    versioning_manager.plugins = plugins
    versioning_manager.transaction_cls = transaction_cls
    versioning_manager.user_cls = user_cls

    engine = create_engine(dns)
    # engine.echo = True

    class Article(Model):
        __tablename__ = "article"
        __versioned__ = copy(options)

        id = sa.Column(sa.Integer, autoincrement=True, primary_key=True)
        name = sa.Column(sa.Unicode(255), nullable=False)
        content = sa.Column(sa.UnicodeText)
        description = sa.Column(sa.UnicodeText)

    class Tag(Model):
        __tablename__ = "tag"
        __versioned__ = copy(options)

        id = sa.Column(sa.Integer, autoincrement=True, primary_key=True)
        article_id = sa.Column(sa.Integer, sa.ForeignKey(Article.id))
        article = sa.orm.relationship(Article, backref="tags")

    sa.orm.configure_mappers()

    connection = engine.connect()

    Model.metadata.create_all(connection)

    Session = sessionmaker(bind=connection)
    session = Session(autoflush=False)
    session.execute("CREATE EXTENSION IF NOT EXISTS hstore")

    Model.metadata.create_all(connection)

    start = time()

    for i in range(20):
        for i in range(20):
            session.add(Article(name="Article", tags=[Tag(), Tag()]))
        session.commit()

    print("Testing with:")
    print("   versioning_strategy=%r" % versioning_strategy)
    print("   property_mod_tracking=%r" % property_mod_tracking)
    print(colored("%r seconds" % (time() - start), "red"))

    Model.metadata.drop_all(connection)

    remove_versioning()
    versioning_manager.reset()

    close_all_sessions()
    session.expunge_all()
    Model.metadata.drop_all(connection)
    engine.dispose()
    connection.close()


setting_variants = {
    "versioning_strategy": [
        "subquery",
        "validity",
    ],
    "property_mod_tracking": [False, True],
}


names = sorted(setting_variants)
combinations = [dict(zip(names, prod)) for prod in it.product(*(setting_variants[name] for name in names))]
for combination in combinations:
    test_versioning(**combination)