beamtous / sandbox
Ritesh Nadhani sandbox. Maybe with all config files like .vimrc and all.
Clone this repository (size: 590.5 KB): HTTPS / SSH
$ hg clone http://bitbucket.org/beamtous/sandbox/
| commit 7: | 8d21aff1e74c |
| parent 6: | 656f91b5ded6 |
| branch: | default |
You can even serialize using xmltramp
Changed (Δ825 bytes):
raw changeset »
mysql_pool/testpool.py (12 lines added, 0 lines removed)
xmltramp/test.py (30 lines added, 7 lines removed)
Up to file-list mysql_pool/testpool.py:
1 |
import time |
|
2 |
from finlib.db import * |
|
3 |
||
4 |
engine = create_engine('mysql://root:viper@localhost/sa-pool', |
|
5 |
pool_recycle=5, echo_pool=True) |
|
6 |
session = Session(bind=engine) |
|
7 |
||
8 |
print session.query(User).get('1') |
|
9 |
||
10 |
time.sleep(11) |
|
11 |
||
12 |
print session.query(User).get('1') |
Up to file-list xmltramp/test.py:
| … | … | @@ -47,7 +47,7 @@ mapper(User, user_table, |
47 |
47 |
}) |
48 |
48 |
mapper(Address, address_table) |
49 |
49 |
|
50 |
engine = create_engine('sqlite:///%s' % DB, echo= |
|
50 |
engine = create_engine('sqlite:///%s' % DB, echo=False) |
|
51 |
51 |
metadata.bind = engine |
52 |
52 |
Session = sessionmaker() |
53 |
53 |
metadata.create_all(engine) |
| … | … | @@ -87,6 +87,16 @@ def deserialize(orm, xmltrampnode): |
87 |
87 |
str(getattr(xmltrampnode, column.name))) |
88 |
88 |
return orm |
89 |
89 |
|
90 |
def serialize(orm, xmltrampnode): |
|
91 |
""" |
|
92 |
Simple method that reads an ORM |
|
93 |
and serializes all the column as element |
|
94 |
node to the node object passed |
|
95 |
""" |
|
96 |
for column in orm.c: |
|
97 |
xmltrampnode[column.name] = str(getattr(orm, column.name)) |
|
98 |
||
99 |
return orm |
|
90 |
100 |
|
91 |
101 |
session = Session(bind=engine) |
92 |
102 |
|
| … | … | @@ -108,15 +118,28 @@ session.commit() |
108 |
118 |
|
109 |
119 |
#Unittest to make sure that the ORM |
110 |
120 |
#was actually saved. |
111 |
session.begin() |
|
112 |
||
113 |
121 |
user = session.query(User).one() |
114 |
122 |
|
115 |
123 |
assert str(user) == str(doc.user.data.username) |
116 |
124 |
assert str(user.address) == str(doc.user.address.address) |
117 |
125 |
|
118 |
print user |
|
119 |
print user.address |
|
126 |
#Test serialization. |
|
127 |
output_xml = """ |
|
128 |
<xml> |
|
129 |
<user> |
|
130 |
<data> |
|
131 |
</data> |
|
132 |
<address> |
|
133 |
</address> |
|
134 |
</user> |
|
135 |
</xml> |
|
136 |
""" |
|
120 |
137 |
|
121 |
session.close() |
|
122 |
||
138 |
output = xmltramp.parse(output_xml) |
|
139 |
user = serialize(user, output.user.data) |
|
140 |
address = serialize(user.address, output.user.address) |
|
141 |
||
142 |
assert output.__repr__(True) == doc.__repr__(True) |
|
143 |
||
144 |
print("Everything seems to be successful...") |
|
145 |