#
# 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 bisect import insort_left, bisect_left
import weakref
import sys
from storm.compat import is_python2, iter_items, iter_zip
from storm.exceptions import PropertyPathError
from storm.info import get_obj_info, get_cls_info
from storm.expr import Column, Undef
from storm.variables import (
Variable, VariableFactory, BoolVariable, IntVariable, FloatVariable,
DecimalVariable, RawStrVariable, UnicodeVariable, DateTimeVariable,
DateVariable, TimeVariable, TimeDeltaVariable, UUIDVariable,
JSONVariable, ListVariable, EnumVariable)
__all__ = [
"Property",
"SimpleProperty",
"Bool",
"Int",
"Float",
"Decimal",
"RawStr",
"Unicode",
"DateTime",
"Date",
"Time",
"TimeDelta",
"UUID",
"Enum",
"JSON",
"List",
"PropertyRegistry",
]
[docs]class Property(object):
def __init__(self, name=None, primary=False,
variable_class=Variable, variable_kwargs={}):
self._name = name
self._primary = primary
self._variable_class = variable_class
self._variable_kwargs = variable_kwargs
def __get__(self, obj, cls=None):
if obj is None:
return self._get_column(cls)
obj_info = get_obj_info(obj)
if cls is None:
# Don't get obj.__class__ because we don't trust it
# (might be proxied or whatever).
cls = obj_info.cls_info.cls
column = self._get_column(cls)
return obj_info.variables[column].get()
def __set__(self, obj, value):
obj_info = get_obj_info(obj)
# Don't get obj.__class__ because we don't trust it
# (might be proxied or whatever).
column = self._get_column(obj_info.cls_info.cls)
obj_info.variables[column].set(value)
def __delete__(self, obj):
obj_info = get_obj_info(obj)
# Don't get obj.__class__ because we don't trust it
# (might be proxied or whatever).
column = self._get_column(obj_info.cls_info.cls)
obj_info.variables[column].delete()
def _detect_attr_name(self, used_cls):
self_id = id(self)
for cls in used_cls.__mro__:
for attr, prop in iter_items(cls.__dict__):
if id(prop) == self_id:
return attr
raise RuntimeError("Property used in an unknown class")
def _get_column(self, cls):
# Cache per-class column values in the class itself, to avoid
# holding a strong reference to it here, and thus rendering
# classes uncollectable in certain situations (e.g. subclasses
# where the property is stored in the base).
try:
# Use class dictionary explicitly to get sensible
# results on subclasses.
column = cls.__dict__["_storm_columns"].get(self)
except KeyError:
cls._storm_columns = {}
column = None
if column is None:
attr = self._detect_attr_name(cls)
if self._name is None:
name = attr
else:
name = self._name
column = PropertyColumn(self, cls, attr, name, self._primary,
self._variable_class,
self._variable_kwargs)
cls._storm_columns[self] = column
return column
class PropertyColumn(Column):
def __init__(self, prop, cls, attr, name, primary,
variable_class, variable_kwargs):
Column.__init__(self, name, cls, primary,
VariableFactory(variable_class, column=self,
validator_attribute=attr,
**variable_kwargs))
self.cls = cls # Used by references
# Copy attributes from the property to avoid one additional
# function call on each access.
for attr in ["__get__", "__set__", "__delete__"]:
setattr(self, attr, getattr(prop, attr))
# This seems to be required on Python 3
if not is_python2:
def __hash__(self):
return id(self)
[docs]class SimpleProperty(Property):
variable_class = None
def __init__(self, name=None, primary=False, **kwargs):
kwargs["value"] = kwargs.pop("default", Undef)
kwargs["value_factory"] = kwargs.pop("default_factory", Undef)
Property.__init__(self, name, primary, self.variable_class, kwargs)
[docs]class Bool(SimpleProperty):
variable_class = BoolVariable
[docs]class Int(SimpleProperty):
variable_class = IntVariable
[docs]class Float(SimpleProperty):
variable_class = FloatVariable
[docs]class Decimal(SimpleProperty):
variable_class = DecimalVariable
[docs]class RawStr(SimpleProperty):
variable_class = RawStrVariable
[docs]class Unicode(SimpleProperty):
variable_class = UnicodeVariable
[docs]class DateTime(SimpleProperty):
variable_class = DateTimeVariable
[docs]class Date(SimpleProperty):
variable_class = DateVariable
[docs]class Time(SimpleProperty):
variable_class = TimeVariable
[docs]class TimeDelta(SimpleProperty):
variable_class = TimeDeltaVariable
[docs]class UUID(SimpleProperty):
variable_class = UUIDVariable
[docs]class JSON(SimpleProperty):
variable_class = JSONVariable
[docs]class List(SimpleProperty):
variable_class = ListVariable
def __init__(self, name=None, **kwargs):
if "default" in kwargs:
raise ValueError("'default' not allowed for List. "
"Use 'default_factory' instead.")
type = kwargs.pop("type", None)
if type is None:
type = Property()
kwargs["item_factory"] = VariableFactory(type._variable_class,
**type._variable_kwargs)
SimpleProperty.__init__(self, name, **kwargs)
[docs]class Enum(SimpleProperty):
"""Enumeration property, allowing used values to differ from stored ones.
For instance::
class Class(Storm):
prop = Enum(map={"one": 1, "two": 2})
obj.prop = "one"
assert obj.prop == "one"
obj.prop = 1 # Raises error.
Another example::
class Class(Storm):
prop = Enum(map={"one": 1, "two": 2}, set_map={"um": 1})
obj.prop = "um"
assert obj.prop is "one"
obj.prop = "one" # Raises error.
"""
variable_class = EnumVariable
def __init__(self, name=None, primary=False, **kwargs):
set_map = dict(kwargs.pop("map"))
get_map = {value: key for key, value in iter_items(set_map)}
if "set_map" in kwargs:
set_map = dict(kwargs.pop("set_map"))
kwargs["get_map"] = get_map
kwargs["set_map"] = set_map
SimpleProperty.__init__(self, name, primary, **kwargs)
[docs]class PropertyRegistry(object):
"""
An object which remembers the Storm properties specified on
classes, and is able to translate names to these properties.
"""
def __init__(self):
self._properties = []
[docs] def get(self, name, namespace=None):
"""Translate a property name path to the actual property.
This method accepts a property name like C{"id"} or C{"Class.id"}
or C{"module.path.Class.id"}, and tries to find a unique
class/property with the given name.
When the C{namespace} argument is given, the registry will be
able to disambiguate names by choosing the one that is closer
to the given namespace. For instance C{get("Class.id", "a.b.c")}
will choose C{a.Class.id} rather than C{d.Class.id}.
"""
key = ".".join(reversed(name.split(".")))+"."
i = bisect_left(self._properties, (key,))
l = len(self._properties)
best_props = []
if namespace is None:
while i < l and self._properties[i][0].startswith(key):
path, prop_ref = self._properties[i]
prop = prop_ref()
if prop is not None:
best_props.append((path, prop))
i += 1
else:
namespace_parts = ("." + namespace).split(".")
best_path_info = (0, sys.maxsize)
while i < l and self._properties[i][0].startswith(key):
path, prop_ref = self._properties[i]
prop = prop_ref()
if prop is None:
i += 1
continue
path_parts = path.split(".")
path_parts.reverse()
common_prefix = 0
for part, ns_part in iter_zip(path_parts, namespace_parts):
if part == ns_part:
common_prefix += 1
else:
break
path_info = (-common_prefix, len(path_parts)-common_prefix)
if path_info < best_path_info:
best_path_info = path_info
best_props = [(path, prop)]
elif path_info == best_path_info:
best_props.append((path, prop))
i += 1
if not best_props:
raise PropertyPathError("Path '%s' matches no known property."
% name)
elif len(best_props) > 1:
paths = [".".join(reversed(path.split(".")[:-1]))
for path, prop in best_props]
raise PropertyPathError("Path '%s' matches multiple "
"properties: %s" %
(name, ", ".join(paths)))
return best_props[0][1]
[docs] def add_class(self, cls):
"""Register properties of C{cls} so that they may be found by C{get()}.
This method may also be used as a class decorator.
"""
suffix = cls.__module__.split(".")
suffix.append(cls.__name__)
suffix.reverse()
suffix = ".%s." % ".".join(suffix)
cls_info = get_cls_info(cls)
for attr in cls_info.attributes:
prop = cls_info.attributes[attr]
prop_ref = weakref.KeyedRef(prop, self._remove, None)
pair = (attr+suffix, prop_ref)
prop_ref.key = pair
insort_left(self._properties, pair)
return cls
[docs] def add_property(self, cls, prop, attr_name):
"""Register property of C{cls} so that it may be found by C{get()}.
"""
suffix = cls.__module__.split(".")
suffix.append(cls.__name__)
suffix.reverse()
suffix = ".%s." % ".".join(suffix)
prop_ref = weakref.KeyedRef(prop, self._remove, None)
pair = (attr_name+suffix, prop_ref)
prop_ref.key = pair
insort_left(self._properties, pair)
[docs] def clear(self):
"""Clean up all properties in the registry.
Used by tests.
"""
del self._properties[:]
def _remove(self, ref):
self._properties.remove(ref.key)
class PropertyPublisherMeta(type):
"""A metaclass that associates subclasses with Storm L{PropertyRegistry}s.
"""
def __init__(self, name, bases, dict):
if not hasattr(self, "_storm_property_registry"):
self._storm_property_registry = PropertyRegistry()
elif hasattr(self, "__storm_table__"):
self._storm_property_registry.add_class(self)