-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathusing_sqlalchemy.py
58 lines (48 loc) · 1.83 KB
/
using_sqlalchemy.py
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
from sqlalchemy import Column, Integer, Index, Text, DateTime, String, Text
from sqlalchemy import Table, create_engine, MetaData
import pdb
class MyDB():
@classmethod
def init(cls, url, echo=True, **kwargs):
cls.engine = create_engine(url, echo=echo, **kwargs)
cls.metadata = MetaData()
cls.metadata.bind = cls.engine
@classmethod
def create_tables(cls):
cls.t1 = Table('t1', cls.metadata,
Column('seqid', Integer, primary_key=True,
autoincrement=True),
Column('serverid', String(50)),
Column('data', Text),
)
cls.t2 = Table('t2', cls.metadata,
Column('seqid', Integer, primary_key=True,
autoincrement=True),
Column('serverid', String(50)),
Column('data', Text),
)
cls.t3 = Table('t3', cls.metadata,
Column('seqid', Integer, primary_key=True,
autoincrement=True),
Column('name', String(50), unique=True),
Column('data', Text),
)
cls.metadata.create_all()
@classmethod
def do_inserts(cls):
count = 0
for table in (cls.t1, cls.t2,):
for i in range(10):
data = "Data-{}".format(count)
count += 1
table.insert(serverid='foo', data=data).execute()
@classmethod
def do_join(cls):
return cls.t1.join(cls.t2, (cls.t1.c.serverid == cls.t2.c.serverid))
@staticmethod
def main(url='sqlite:///:memory:'):
MyDB.init(url)
MyDB.create_tables()
MyDB.do_inserts()
pdb.set_trace()
print(MyDB.do_join())