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
|
"""
"polymorphic" associations, ala SQLAlchemy.
This example generalizes the function in poly_assoc_pk.py into a
function "association" which creates a new polymorphic association
"interface".
"""
from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey
from sqlalchemy.orm import mapper, relationship, create_session, class_mapper
metadata = MetaData('sqlite://')
def association(cls, table):
"""create an association 'interface'."""
interface_name = table.name
attr_name = "%s_rel" % interface_name
metadata = table.metadata
association_table = Table("%s_associations" % interface_name, metadata,
Column('assoc_id', Integer, primary_key=True),
Column('type', String(50), nullable=False)
)
class GenericAssoc(object):
def __init__(self, name):
self.type = name
def interface(cls, name, uselist=True):
mapper = class_mapper(cls)
table = mapper.local_table
mapper.add_property(attr_name, relationship(GenericAssoc, backref='_backref_%s' % table.name))
if uselist:
# list based property decorator
def get(self):
if getattr(self, attr_name) is None:
setattr(self, attr_name, GenericAssoc(table.name))
return getattr(self, attr_name).targets
setattr(cls, name, property(get))
else:
# scalar based property decorator
def get(self):
return getattr(self, attr_name).targets[0]
def set(self, value):
if getattr(self, attr_name) is None:
setattr(self, attr_name, GenericAssoc(table.name))
getattr(self, attr_name).targets = [value]
setattr(cls, name, property(get, set))
setattr(cls, 'member', property(lambda self: getattr(self.association, '_backref_%s' % self.association.type)))
mapper(GenericAssoc, association_table, properties={
'targets':relationship(cls, backref='association'),
})
return interface
#######
# addresses table
addresses = Table("addresses", metadata,
Column('id', Integer, primary_key=True),
Column('assoc_id', None, ForeignKey('addresses_associations.assoc_id')),
Column('street', String(100)),
Column('city', String(50)),
Column('country', String(50))
)
class Address(object):
pass
# create "addressable" association
addressable = association(Address, addresses)
mapper(Address, addresses)
######
# sample # 1, users
users = Table("users", metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50), nullable=False),
Column('assoc_id', None, ForeignKey('addresses_associations.assoc_id'))
)
class User(object):
pass
mapper(User, users)
# use the association
addressable(User, 'addresses', uselist=True)
######
# sample # 2, orders
orders = Table("orders", metadata,
Column('id', Integer, primary_key=True),
Column('description', String(50), nullable=False),
Column('assoc_id', None, ForeignKey('addresses_associations.assoc_id'))
)
class Order(object):
pass
mapper(Order, orders)
addressable(Order, 'address', uselist=False)
######
# use it !
metadata.create_all()
u1 = User()
u1.name = 'bob'
o1 = Order()
o1.description = 'order 1'
a1 = Address()
u1.addresses.append(a1)
a1.street = '123 anywhere street'
a2 = Address()
u1.addresses.append(a2)
a2.street = '345 orchard ave'
o1.address = Address()
o1.address.street = '444 park ave.'
sess = create_session()
sess.add(u1)
sess.add(o1)
sess.flush()
sess.expunge_all()
# query objects, get their addresses
bob = sess.query(User).filter_by(name='bob').one()
assert [s.street for s in bob.addresses] == ['123 anywhere street', '345 orchard ave']
order = sess.query(Order).filter_by(description='order 1').one()
assert order.address.street == '444 park ave.'
# query from Address to members
for address in sess.query(Address).all():
print "Street", address.street, "Member", address.member
|
