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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
|
"""Mapping a polymorphic-valued vertical table as a dictionary.
Builds upon the dictlike.py example to also add differently typed
columns to the "fact" table, e.g.::
Table('properties', metadata
Column('owner_id', Integer, ForeignKey('owner.id'),
primary_key=True),
Column('key', UnicodeText),
Column('type', Unicode(16)),
Column('int_value', Integer),
Column('char_value', UnicodeText),
Column('bool_value', Boolean),
Column('decimal_value', Numeric(10,2)))
For any given properties row, the value of the 'type' column will point to the
'_value' column active for that row.
This example approach uses exactly the same dict mapping approach as the
'dictlike' example. It only differs in the mapping for vertical rows. Here,
we'll use a @hybrid_property to build a smart '.value' attribute that wraps up
reading and writing those various '_value' columns and keeps the '.type' up to
date.
"""
from sqlalchemy import and_
from sqlalchemy import Boolean
from sqlalchemy import case
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import literal_column
from sqlalchemy import null
from sqlalchemy import or_
from sqlalchemy import String
from sqlalchemy import Unicode
from sqlalchemy import UnicodeText
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.interfaces import PropComparator
from .dictlike import ProxiedDictMixin
class PolymorphicVerticalProperty:
"""A key/value pair with polymorphic value storage.
The class which is mapped should indicate typing information
within the "info" dictionary of mapped Column objects; see
the AnimalFact mapping below for an example.
"""
def __init__(self, key, value=None):
self.key = key
self.value = value
@hybrid_property
def value(self):
fieldname, discriminator = self.type_map[self.type]
if fieldname is None:
return None
else:
return getattr(self, fieldname)
@value.setter
def value(self, value):
py_type = type(value)
fieldname, discriminator = self.type_map[py_type]
self.type = discriminator
if fieldname is not None:
setattr(self, fieldname, value)
@value.deleter
def value(self):
self._set_value(None)
@value.comparator
class value(PropComparator):
"""A comparator for .value, builds a polymorphic comparison
via CASE."""
def __init__(self, cls):
self.cls = cls
def _case(self):
pairs = set(self.cls.type_map.values())
whens = [
(
literal_column("'%s'" % discriminator),
cast(getattr(self.cls, attribute), String),
)
for attribute, discriminator in pairs
if attribute is not None
]
return case(*whens, value=self.cls.type, else_=null())
def __eq__(self, other):
return self._case() == cast(other, String)
def __ne__(self, other):
return self._case() != cast(other, String)
def __repr__(self):
return "<%s %r=%r>" % (self.__class__.__name__, self.key, self.value)
@event.listens_for(
PolymorphicVerticalProperty, "mapper_configured", propagate=True
)
def on_new_class(mapper, cls_):
"""Look for Column objects with type info in them, and work up
a lookup table."""
info_dict = {}
info_dict[type(None)] = (None, "none")
info_dict["none"] = (None, "none")
for k in mapper.c.keys():
col = mapper.c[k]
if "type" in col.info:
python_type, discriminator = col.info["type"]
info_dict[python_type] = (k, discriminator)
info_dict[discriminator] = (k, discriminator)
cls_.type_map = info_dict
if __name__ == "__main__":
Base = declarative_base()
class AnimalFact(PolymorphicVerticalProperty, Base):
"""A fact about an animal."""
__tablename__ = "animal_fact"
animal_id = Column(ForeignKey("animal.id"), primary_key=True)
key = Column(Unicode(64), primary_key=True)
type = Column(Unicode(16))
# add information about storage for different types
# in the info dictionary of Columns
int_value = Column(Integer, info={"type": (int, "integer")})
char_value = Column(UnicodeText, info={"type": (str, "string")})
boolean_value = Column(Boolean, info={"type": (bool, "boolean")})
class Animal(ProxiedDictMixin, Base):
"""an Animal"""
__tablename__ = "animal"
id = Column(Integer, primary_key=True)
name = Column(Unicode(100))
facts = relationship(
"AnimalFact", collection_class=attribute_keyed_dict("key")
)
_proxied = association_proxy(
"facts",
"value",
creator=lambda key, value: AnimalFact(key=key, value=value),
)
def __init__(self, name):
self.name = name
def __repr__(self):
return "Animal(%r)" % self.name
@classmethod
def with_characteristic(self, key, value):
return self.facts.any(key=key, value=value)
engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)
session = Session(engine)
stoat = Animal("stoat")
stoat["color"] = "red"
stoat["cuteness"] = 7
stoat["weasel-like"] = True
session.add(stoat)
session.commit()
critter = session.query(Animal).filter(Animal.name == "stoat").one()
print(critter["color"])
print(critter["cuteness"])
print("changing cuteness value and type:")
critter["cuteness"] = "very cute"
session.commit()
marten = Animal("marten")
marten["cuteness"] = 5
marten["weasel-like"] = True
marten["poisonous"] = False
session.add(marten)
shrew = Animal("shrew")
shrew["cuteness"] = 5
shrew["weasel-like"] = False
shrew["poisonous"] = True
session.add(shrew)
session.commit()
q = session.query(Animal).filter(
Animal.facts.any(
and_(AnimalFact.key == "weasel-like", AnimalFact.value == True)
)
)
print("weasel-like animals", q.all())
q = session.query(Animal).filter(
Animal.with_characteristic("weasel-like", True)
)
print("weasel-like animals again", q.all())
q = session.query(Animal).filter(
Animal.with_characteristic("poisonous", False)
)
print("animals with poisonous=False", q.all())
q = session.query(Animal).filter(
or_(
Animal.with_characteristic("poisonous", False),
~Animal.facts.any(AnimalFact.key == "poisonous"),
)
)
print("non-poisonous animals", q.all())
q = session.query(Animal).filter(Animal.facts.any(AnimalFact.value == 5))
print("any animal with a .value of 5", q.all())
|
