/usr/share/pyshared/storm/info.py is in python-storm 0.19-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 246 247 248 | #
# Copyright (c) 2006, 2007 Canonical
#
# Written by Gustavo Niemeyer <gustavo@niemeyer.net>
#
# This file is part of Storm Object Relational Mapper.
#
# Storm is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation; either version 2.1 of
# the License, or (at your option) any later version.
#
# Storm is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from weakref import ref
from storm.exceptions import ClassInfoError
from storm.expr import Column, Desc, TABLE
from storm.expr import compile, Table
from storm.event import EventSystem
from storm import Undef, has_cextensions
__all__ = ["get_obj_info", "set_obj_info", "get_cls_info",
"ClassInfo", "ObjectInfo", "ClassAlias"]
def get_obj_info(obj):
try:
return obj.__storm_object_info__
except AttributeError:
# Instantiate ObjectInfo first, so that it breaks gracefully,
# in case the object isn't a storm object.
obj_info = ObjectInfo(obj)
return obj.__dict__.setdefault("__storm_object_info__", obj_info)
def set_obj_info(obj, obj_info):
obj.__dict__["__storm_object_info__"] = obj_info
def get_cls_info(cls):
if "__storm_class_info__" in cls.__dict__:
# Can't use attribute access here, otherwise subclassing won't work.
return cls.__dict__["__storm_class_info__"]
else:
cls.__storm_class_info__ = ClassInfo(cls)
return cls.__storm_class_info__
class ClassInfo(dict):
"""Persistent storm-related information of a class.
The following attributes are defined:
@ivar table: Expression from where columns will be looked up.
@ivar cls: Class which should be used to build objects.
@ivar columns: Tuple of column properties found in the class.
@ivar primary_key: Tuple of column properties used to form the primary key
@ivar primary_key_pos: Position of primary_key items in the columns tuple.
"""
def __init__(self, cls):
self.table = getattr(cls, "__storm_table__", None)
if self.table is None:
raise ClassInfoError("%s.__storm_table__ missing" % repr(cls))
self.cls = cls
if isinstance(self.table, basestring):
self.table = Table(self.table)
pairs = []
for attr in dir(cls):
column = getattr(cls, attr, None)
if isinstance(column, Column):
pairs.append((attr, column))
pairs.sort()
self.columns = tuple(pair[1] for pair in pairs)
self.attributes = dict(pairs)
storm_primary = getattr(cls, "__storm_primary__", None)
if storm_primary is not None:
if type(storm_primary) is not tuple:
storm_primary = (storm_primary,)
self.primary_key = tuple(self.attributes[attr]
for attr in storm_primary)
else:
primary = []
primary_attrs = {}
for attr, column in pairs:
if column.primary != 0:
if column.primary in primary_attrs:
raise ClassInfoError(
"%s has two columns with the same primary id: "
"%s and %s" %
(repr(cls), attr, primary_attrs[column.primary]))
primary.append((column.primary, column))
primary_attrs[column.primary] = attr
primary.sort()
self.primary_key = tuple(column for i, column in primary)
if not self.primary_key:
raise ClassInfoError("%s has no primary key information" %
repr(cls))
# columns have __eq__ implementations that do things we don't want - we
# want to look these up in a dict and use identity semantics
id_positions = dict((id(column), i)
for i, column in enumerate(self.columns))
self.primary_key_idx = dict((id(column), i)
for i, column in
enumerate(self.primary_key))
self.primary_key_pos = tuple(id_positions[id(column)]
for column in self.primary_key)
__order__ = getattr(cls, "__storm_order__", None)
if __order__ is None:
self.default_order = Undef
else:
if type(__order__) is not tuple:
__order__ = (__order__,)
self.default_order = []
for item in __order__:
if isinstance(item, basestring):
if item.startswith("-"):
prop = Desc(getattr(cls, item[1:]))
else:
prop = getattr(cls, item)
else:
prop = item
self.default_order.append(prop)
def __eq__(self, other):
return self is other
def __ne__(self, other):
return self is not other
class ObjectInfo(dict):
__hash__ = object.__hash__
# For get_obj_info(), an ObjectInfo is its own obj_info.
__storm_object_info__ = property(lambda self: self)
def __init__(self, obj):
# FASTPATH This method is part of the fast path. Be careful when
# changing it (try to profile any changes).
# First thing, try to create a ClassInfo for the object's class.
# This ensures that obj is the kind of object we expect.
self.cls_info = get_cls_info(type(obj))
self.set_obj(obj)
self.event = event = EventSystem(self)
self.variables = variables = {}
for column in self.cls_info.columns:
variables[column] = \
column.variable_factory(column=column,
event=event,
validator_object_factory=self.get_obj)
self.primary_vars = tuple(variables[column]
for column in self.cls_info.primary_key)
def __eq__(self, other):
return self is other
def __ne__(self, other):
return self is not other
def set_obj(self, obj):
self._ref = ref(obj, self._emit_object_deleted)
def get_obj(self):
return self._ref()
def _emit_object_deleted(self, obj_ref):
self.event.emit("object-deleted")
def checkpoint(self):
for variable in self.variables.itervalues():
variable.checkpoint()
if has_cextensions:
from storm.cextensions import ObjectInfo, get_obj_info
class ClassAlias(object):
"""Create a named alias for a Storm class for use in queries.
This is useful basically when the SQL 'AS' feature is desired in code using
Storm queries.
ClassAliases which are explicitly named (i.e., when 'name' is passed) are
cached for as long as the class exists, such that the alias returned from
C{ClassAlias(Foo, 'foo_alias')} will be the same object no matter how many
times it's called.
@param cls: The class to create the alias of.
@param name: If provided, specify the name of the alias to create.
"""
alias_count = 0
def __new__(self_cls, cls, name=Undef):
if name is Undef:
use_cache = False
ClassAlias.alias_count += 1
name = "_%x" % ClassAlias.alias_count
else:
use_cache = True
cache = cls.__dict__.get("_storm_alias_cache")
if cache is None:
cls._storm_alias_cache = {}
elif name in cache:
return cache[name]
alias_cls = type(cls.__name__ + "Alias", (self_cls,),
{"__storm_table__": name})
alias_cls.__bases__ = (cls, self_cls)
alias_cls_info = get_cls_info(alias_cls)
alias_cls_info.cls = cls
if use_cache:
cls._storm_alias_cache[name] = alias_cls
return alias_cls
@compile.when(type)
def compile_type(compile, expr, state):
cls_info = get_cls_info(expr)
table = compile(cls_info.table, state)
if state.context is TABLE and issubclass(expr, ClassAlias):
return "%s AS %s" % (compile(cls_info.cls, state), table)
return table
|