-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathm2m.py~
67 lines (56 loc) · 1.72 KB
/
m2m.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
59
60
61
62
63
64
65
66
67
# -*- coding: utf-8 -*-
from sqlalchemy import *
from sqlalchemy.ext.declarative import *
from sqlalchemy.orm import *
from sqlalchemy_utils import *
base = declarative_base()
assoc = Table('assoc', base.metadata,
Column('aform_id', Integer, ForeignKey('AForm.id')),
Column('bform_id', Integer, ForeignKey('BForm.id')),
Column('tag_id', Integer, ForeignKey('Tag.id'))
)
class Tag(base):
__tablename__ = 'Tag'
id = Column(Integer, primary_key = True)
name = Column(String)
aform = relationship('AForm', secondary = assoc, back_populates = 'atag')
bform = relationship('BForm', secondary = assoc, back_populates = 'btag')
class Form(AbstractConcreteBase, base):
id = Column(Integer, primary_key = True)
amount = Column(Integer)
@declared_attr
def __tablename__(cls):
return cls.__name__
@declared_attr
def __mapper_args__(cls):
return {
'polymorphic_identity': cls.__name__,
'concrete':True
}
class AForm(Form, base):
atag = relationship('Tag', secondary = assoc, back_populates = 'aform')
class BForm(Form, base):
btag = relationship('Tag', secondary = assoc, back_populates = 'bform')
db_uri = 'sqlite:////tmp/m2m.sqlite'
engine = create_engine(db_uri, echo =True)
if database_exists(engine.url):
drop_database(db_uri)
create_database(engine.url)
base.metadata.create_all(bind = engine)
Session = sessionmaker(bind = engine)
session = Session()
a = AForm(amount = 100)
atag = Tag(name = 'booked')
a.atag.append(atag)
session.add(a)
b = BForm(amount = 200)
btag = Tag(name = 'canceled')
b.btag.append(btag)
session.add(b)
session.commit()
session.query(AForm).all()
forms=session.query(Form.amount, Form).all()
for f in forms:
print(f)
# print(f.atag[0].name)
# print(f.btag[0].name)