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
118
119
120
121
|
from testbase import PersistTest
import testbase
import unittest, sys, datetime
import sqlalchemy.databases.sqlite as sqllite
db = testbase.db
from sqlalchemy import *
class QueryTest(PersistTest):
def setUpAll(self):
global users
users = Table('query_users', db,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
redefine = True
)
users.create()
def setUp(self):
self.users = users
def tearDown(self):
self.users.delete().execute()
def tearDownAll(self):
global users
users.drop()
def testinsert(self):
self.users.insert().execute(user_id = 7, user_name = 'jack')
print repr(self.users.select().execute().fetchall())
def testupdate(self):
self.users.insert().execute(user_id = 7, user_name = 'jack')
print repr(self.users.select().execute().fetchall())
self.users.update(self.users.c.user_id == 7).execute(user_name = 'fred')
print repr(self.users.select().execute().fetchall())
def testselectdate(self):
users_with_date = Table('query_users_with_date', db,
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
Column('user_date', DateTime),
redefine = True
)
users_with_date.create()
c = db.connection()
users_with_date.insert().execute(user_id = 7, user_name = 'jack', user_date=datetime.datetime(2005,11,10))
users_with_date.insert().execute(user_id = 8, user_name = 'roy', user_date=datetime.datetime(2005,11,10, 11,52,35))
users_with_date.insert().execute(user_id = 9, user_name = 'foo', user_date=datetime.datetime(2005,11,10, 11,52,35, 54839))
users_with_date.insert().execute(user_id = 10, user_name = 'colber', user_date=None)
print repr(users_with_date.select().execute().fetchall())
users_with_date.drop()
def testdefaults(self):
x = {'x':0}
def mydefault():
x['x'] += 1
return x['x']
# select "count(1)" from the DB which returns different results
# on different DBs
f = select([func.count(1)], engine=db).execute().fetchone()[0]
t = Table('default_test1', db,
Column('col1', Integer, primary_key=True, default=mydefault),
Column('col2', String(20), default="imthedefault"),
Column('col3', Integer, default=func.count(1)),
)
t.create()
try:
t.insert().execute()
t.insert().execute()
t.insert().execute()
l = t.select().execute()
self.assert_(l.fetchall() == [(1, 'imthedefault', f), (2, 'imthedefault', f), (3, 'imthedefault', f)])
finally:
t.drop()
def testdelete(self):
c = db.connection()
self.users.insert().execute(user_id = 7, user_name = 'jack')
self.users.insert().execute(user_id = 8, user_name = 'fred')
print repr(self.users.select().execute().fetchall())
self.users.delete(self.users.c.user_name == 'fred').execute()
print repr(self.users.select().execute().fetchall())
def testtransaction(self):
def dostuff():
self.users.insert().execute(user_id = 7, user_name = 'john')
self.users.insert().execute(user_id = 8, user_name = 'jack')
db.transaction(dostuff)
print repr(self.users.select().execute().fetchall())
def testselectlimit(self):
self.users.insert().execute(user_id=1, user_name='john')
self.users.insert().execute(user_id=2, user_name='jack')
self.users.insert().execute(user_id=3, user_name='ed')
self.users.insert().execute(user_id=4, user_name='wendy')
self.users.insert().execute(user_id=5, user_name='laura')
self.users.insert().execute(user_id=6, user_name='ralph')
self.users.insert().execute(user_id=7, user_name='fido')
r = self.users.select(limit=3).execute().fetchall()
self.assert_(r == [(1, 'john'), (2, 'jack'), (3, 'ed')])
r = self.users.select(limit=3, offset=2).execute().fetchall()
self.assert_(r==[(3, 'ed'), (4, 'wendy'), (5, 'laura')])
r = self.users.select(offset=5).execute().fetchall()
self.assert_(r==[(6, 'ralph'), (7, 'fido')])
if __name__ == "__main__":
testbase.main()
|
