-
Notifications
You must be signed in to change notification settings - Fork 30
/
itests.txt
210 lines (143 loc) · 5.17 KB
/
itests.txt
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
==================
SQLAlchemy Dialect
==================
>>> connection = engine.connect()
Using the connection to execute a select statement::
>>> result = connection.execute('select name from locations order by name')
>>> result.rowcount
13
>>> result.first()
('Aldebaran',)
Using the ORM to query the locations::
>>> locations = session.query(Location).order_by('name')
>>> [l.name for l in locations if l is not None][:2]
['Aldebaran', 'Algol']
With limit and offset::
>>> locations = session.query(Location).order_by('name').offset(1).limit(2)
>>> [l.name for l in locations if l is not None]
['Algol', 'Allosimanius Syneca']
With filter::
>>> location = session.query(Location).filter_by(name='Algol').one()
>>> location.name
'Algol'
Order by::
>>> locations = session.query(Location).filter(Location.name != None).order_by(sa.desc(Location.name))
>>> locations = locations.limit(2)
>>> [l.name for l in locations]
['Outer Eastern Rim', 'North West Ripple']
Insert a new location::
>>> location = Location()
>>> location.name = 'Earth'
>>> location.kind = 'Planet'
>>> location.flag = True
>>> session.add(location)
>>> session.flush()
Currently refresh option is missing, therefore sleep for now::
>>> from time import sleep
>>> sleep(1)
Inserted location is available::
>>> location = session.query(Location).filter_by(name='Earth').one()
>>> location.name
'Earth'
Retrieve the location from the database::
>>> session.refresh(location)
>>> location.name
'Earth'
Date should have been set at the insert due to default value via python method::
>>> from datetime import datetime
>>> now = datetime.utcnow()
>>> dt = location.date
>>> dt.year == now.year
True
>>> dt.month == now.month
True
>>> dt.day == now.day
True
>>> (now - location.datetime).seconds < 4
True
Verify the return type of date and datetime::
>>> type(location.date)
<class 'datetime.date'>
>>> type(location.datetime)
<class 'datetime.datetime'>
the location also has a date and datetime property which both are nullable and
aren't set when the row is inserted as there is no default method::
>>> location.nullable_datetime is None
True
>>> location.nullable_date is None
True
The datetime and date can be set using a update statement::
>>> location.nullable_date = datetime.today()
>>> location.nullable_datetime = datetime.utcnow()
>>> session.flush()
>>> sleep(1.1) # wait for index refresh
Boolean values get set natively::
>>> location.flag
True
Reload the object from the db::
>>> session.refresh(location)
And verify that the date and datetime was persisted::
>>> location.nullable_datetime is not None
True
>>> location.nullable_date is not None
True
Update one Location via raw sql::
>>> result = connection.execute("update locations set kind='Heimat' where name='Earth'")
>>> result.rowcount
1
Update multiple Locations::
>>> for x in range(10):
... loc = Location()
... loc.name = 'Ort %d' % x
... loc.kind = 'Update'
... session.add(loc)
... session.flush()
>>> sleep(2) # give crate some time to settle
>>> result = connection.execute("update locations set flag=true where kind='Update'")
>>> result.rowcount
10
Check that number of affected documents of update without ``where-clause`` matches number of all
documents in the table::
>>> result = connection.execute(u"update locations set kind='Überall'")
>>> result.rowcount == connection.execute("select * from locations limit 100").rowcount
True
>>> session.commit()
>>> sleep(2) # give crate some time to settle
Test that objects can be used as list too::
>>> location = session.query(Location).filter_by(name='Folfanga').one()
>>> location.details = [{'size': 'huge'}, {'clima': 'cold'}]
>>> session.commit()
>>> session.refresh(location)
>>> location.details
[{'size': 'huge'}, {'clima': 'cold'}]
Update the clima::
>>> location.details[1] = {'clima': 'hot'}
>>> session.commit()
>>> session.refresh(location)
>>> location.details
[{'size': 'huge'}, {'clima': 'hot'}]
Reset the clima::
>>> location.details = []
>>> session.commit()
>>> session.refresh(location)
>>> location.details
[]
test updated nested dict::
>>> from crate.client.sqlalchemy.types import Craty
>>> class Character(Base):
... __tablename__ = 'characters'
... id = sa.Column(sa.String, primary_key=True)
... details = sa.Column(Craty)
>>> char = Character(id='1234id')
>>> char.details = {"name": {"first": "Arthur", "last": "Dent"}}
>>> session.add(char)
>>> session.commit()
>>> char = session.query(Character).filter_by(id='1234id').one()
>>> char.details['name']['first'] = 'Trillian'
>>> char.details['size'] = 45
>>> session.commit()
>>> sleep(1.1) # wait for index refresh
>>> session.refresh(char)
>>> import pprint
>>> pprint.pprint(char.details)
{'name': {'first': 'Trillian', 'last': 'Dent'}, 'size': 45}