File: massload.py

package info (click to toggle)
sqlalchemy 0.6.3-3%2Bsqueeze1
  • links: PTS, VCS
  • area: main
  • in suites: squeeze
  • size: 10,744 kB
  • ctags: 15,132
  • sloc: python: 93,431; ansic: 787; makefile: 137; xml: 17
file content (63 lines) | stat: -rw-r--r-- 2,161 bytes parent folder | download
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
import time
#import sqlalchemy.orm.attributes as attributes
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.test import *

"""

we are testing session.expunge() here, also that the attributes and unitofwork
packages dont keep dereferenced stuff hanging around.

for best results, dont run with sqlite :memory: database, and keep an eye on
top while it runs
"""

NUM = 2500

class LoadTest(TestBase, AssertsExecutionResults):
    @classmethod
    def setup_class(cls):
        global items, meta
        meta = MetaData(testing.db)
        items = Table('items', meta,
            Column('item_id', Integer, primary_key=True),
            Column('value', String(100)))
        items.create()
    @classmethod
    def teardown_class(cls):
        items.drop()
    def setup(self):
        for x in range(1,NUM/500+1):
            l = []
            for y in range(x*500-500 + 1, x*500 + 1):
                l.append({'item_id':y, 'value':'this is item #%d' % y})
            items.insert().execute(*l)

    def testload(self):
        class Item(object):pass

        m = mapper(Item, items)
        sess = create_session()
        now = time.time()
        query = sess.query(Item)
        for x in range (1,NUM/100):
            # this is not needed with cpython which clears non-circular refs immediately
            #gc_collect()
            l = query.filter(items.c.item_id.between(x*100 - 100 + 1, x*100)).all()
            assert len(l) == 100
            print "loaded ", len(l), " items "
            # modifying each object will ensure that the objects get placed in the "dirty" list
            # and will hang around until expunged
            #for a in l:
            #    a.value = 'changed...'
            #assert len(objectstore.get_session().dirty) == len(l)
            #assert len(objectstore.get_session().identity_map) == len(l)
            #assert len(attributes.managed_attributes) == len(l)
            #print len(objectstore.get_session().dirty)
            #print len(objectstore.get_session().identity_map)
            #objectstore.expunge(*l)
        total = time.time() -now
        print "total time ", total