series as well. For changes that are specific to 1.0 with an emphasis
on compatibility concerns, see :doc:`/changelog/migration_10`.
+ .. change::
+ :tags: bug, declarative
+ :tickets: 2670
+
+ A relationship set up with :class:`.declared_attr` on
+ a :class:`.AbstractConcreteBase` base class will now be configured
+ on the abstract base mapping automatically, in addition to being
+ set up on descendant concrete classes as usual.
+
+ .. seealso::
+
+ :ref:`feature_3150`
+
+ .. change::
+ :tags: feature, declarative
+ :tickets: 3150
+
+ The :class:`.declared_attr` construct has newly improved
+ behaviors and features in conjunction with declarative. The
+ decorated function will now have access to the final column
+ copies present on the local mixin when invoked, and will also
+ be invoked exactly once for each mapped class, the returned result
+ being memoized. A new modifier :attr:`.declared_attr.cascading`
+ is added as well.
+
+ .. seealso::
+
+ :ref:`feature_3150`
+
.. change::
:tags: feature, ext
:tickets: 3210
undergoing maintenance releases as of May, 2014,
and SQLAlchemy version 1.0, as of yet unreleased.
- Document last updated: September 7, 2014
+ Document last updated: September 25, 2014
Introduction
============
:ticket:`3177`
+.. _feature_3150:
+
+Improvements to declarative mixins, ``@declared_attr`` and related features
+----------------------------------------------------------------------------
+
+The declarative system in conjunction with :class:`.declared_attr` has been
+overhauled to support new capabilities.
+
+A function decorated with :class:`.declared_attr` is now called only **after**
+any mixin-based column copies are generated. This means the function can
+call upon mixin-established columns and will receive a reference to the correct
+:class:`.Column` object::
+
+ class HasFooBar(object):
+ foobar = Column(Integer)
+
+ @declared_attr
+ def foobar_prop(cls):
+ return column_property('foobar: ' + cls.foobar)
+
+ class SomeClass(HasFooBar, Base):
+ __tablename__ = 'some_table'
+ id = Column(Integer, primary_key=True)
+
+Above, ``SomeClass.foobar_prop`` will be invoked against ``SomeClass``,
+and ``SomeClass.foobar`` will be the final :class:`.Column` object that is
+to be mapped to ``SomeClass``, as opposed to the non-copied object present
+directly on ``HasFooBar``, even though the columns aren't mapped yet.
+
+The :class:`.declared_attr` function now **memoizes** the value
+that's returned on a per-class basis, so that repeated calls to the same
+attribute will return the same value. We can alter the example to illustrate
+this::
+
+ class HasFooBar(object):
+ @declared_attr
+ def foobar(cls):
+ return Column(Integer)
+
+ @declared_attr
+ def foobar_prop(cls):
+ return column_property('foobar: ' + cls.foobar)
+
+ class SomeClass(HasFooBar, Base):
+ __tablename__ = 'some_table'
+ id = Column(Integer, primary_key=True)
+
+Previously, ``SomeClass`` would be mapped with one particular copy of
+the ``foobar`` column, but the ``foobar_prop`` by calling upon ``foobar``
+a second time would produce a different column. The value of
+``SomeClass.foobar`` is now memoized during declarative setup time, so that
+even before the attribute is mapped by the mapper, the interim column
+value will remain consistent no matter how many times the
+:class:`.declared_attr` is called upon.
+
+The two behaviors above should help considerably with declarative definition
+of many types of mapper properties that derive from other attributes, where
+the :class:`.declared_attr` function is called upon from other
+:class:`.declared_attr` functions locally present before the class is
+actually mapped.
+
+For a pretty slim edge case where one wishes to build a declarative mixin
+that establishes distinct columns per subclass, a new modifier
+:attr:`.declared_attr.cascading` is added. With this modifier, the
+decorated function will be invoked individually for each class in the
+mapped inheritance hierarchy. While this is already the behavior for
+special attributes such as ``__table_args__`` and ``__mapper_args__``,
+for columns and other properties the behavior by default assumes that attribute
+is affixed to the base class only, and just inherited from subclasses.
+With :attr:`.declared_attr.cascading`, individual behaviors can be
+applied::
+
+ class HasSomeAttribute(object):
+ @declared_attr.cascading
+ def some_id(cls):
+ if has_inherited_table(cls):
+ return Column(ForeignKey('myclass.id'), primary_key=True)
+ else:
+ return Column(Integer, primary_key=True)
+
+ return Column('id', Integer, primary_key=True)
+
+ class MyClass(HasSomeAttribute, Base):
+ ""
+ # ...
+
+ class MySubClass(MyClass):
+ ""
+ # ...
+
+.. seealso::
+
+ :ref:`mixin_inheritance_columns`
+
+Finally, the :class:`.AbstractConcreteBase` class has been reworked
+so that a relationship or other mapper property can be set up inline
+on the abstract base::
+
+ from sqlalchemy import Column, Integer, ForeignKey
+ from sqlalchemy.orm import relationship
+ from sqlalchemy.ext.declarative import (declarative_base, declared_attr,
+ AbstractConcreteBase)
+
+ Base = declarative_base()
+
+ class Something(Base):
+ __tablename__ = u'something'
+ id = Column(Integer, primary_key=True)
+
+
+ class Abstract(AbstractConcreteBase, Base):
+ id = Column(Integer, primary_key=True)
+
+ @declared_attr
+ def something_id(cls):
+ return Column(ForeignKey(Something.id))
+
+ @declared_attr
+ def something(cls):
+ return relationship(Something)
+
+
+ class Concrete(Abstract):
+ __tablename__ = u'cca'
+ __mapper_args__ = {'polymorphic_identity': 'cca', 'concrete': True}
+
+
+The above mapping will set up a table ``cca`` with both an ``id`` and
+a ``something_id`` column, and ``Concrete`` will also have a relationship
+``something``. The new feature is that ``Abstract`` will also have an
+independently configured relationship ``something`` that builds against
+the polymorphic union of the base.
+
+:ticket:`3150` :ticket:`2670` :ticket:`3149` :ticket:`2952` :ticket:`3050`
.. _bug_3188:
.. autofunction:: as_declarative
.. autoclass:: declared_attr
+ :members:
.. autofunction:: sqlalchemy.ext.declarative.api._declarative_constructor
Columns generated by :class:`~.declared_attr` can also be
referenced by ``__mapper_args__`` to a limited degree, currently
-by ``polymorphic_on`` and ``version_id_col``, by specifying the
-classdecorator itself into the dictionary - the declarative extension
+by ``polymorphic_on`` and ``version_id_col``; the declarative extension
will resolve them at class construction time::
class MyMixin:
id = Column(Integer, primary_key=True)
-
Mixing in Relationships
~~~~~~~~~~~~~~~~~~~~~~~
__tablename__ = 'target'
id = Column(Integer, primary_key=True)
+
Using Advanced Relationship Arguments (e.g. ``primaryjoin``, etc.)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
class Something(SomethingMixin, Base):
__tablename__ = "something"
+The :func:`.column_property` or other construct may refer
+to other columns from the mixin. These are copied ahead of time before
+the :class:`.declared_attr` is invoked::
+
+ class SomethingMixin(object):
+ x = Column(Integer)
+
+ y = Column(Integer)
+
+ @declared_attr
+ def x_plus_y(cls):
+ return column_property(cls.x + cls.y)
+
+
+.. versionchanged:: 1.0.0 mixin columns are copied to the final mapped class
+ so that :class:`.declared_attr` methods can access the actual column
+ that will be mapped.
+
Mixing in Association Proxy and Other Attributes
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Controlling table inheritance with mixins
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-The ``__tablename__`` attribute in conjunction with the hierarchy of
-classes involved in a declarative mixin scenario controls what type of
-table inheritance, if any,
-is configured by the declarative extension.
+The ``__tablename__`` attribute may be used to provide a function that
+will determine the name of the table used for each class in an inheritance
+hierarchy, as well as whether a class has its own distinct table.
-If the ``__tablename__`` is computed by a mixin, you may need to
-control which classes get the computed attribute in order to get the
-type of table inheritance you require.
+This is achieved using the :class:`.declared_attr` indicator in conjunction
+with a method named ``__tablename__()``. Declarative will always
+invoke :class:`.declared_attr` for the special names
+``__tablename__``, ``__mapper_args__`` and ``__table_args__``
+function **for each mapped class in the hierarchy**. The function therefore
+needs to expect to receive each class individually and to provide the
+correct answer for each.
-For example, if you had a mixin that computes ``__tablename__`` but
-where you wanted to use that mixin in a single table inheritance
-hierarchy, you can explicitly specify ``__tablename__`` as ``None`` to
-indicate that the class should not have a table mapped::
+For example, to create a mixin that gives every class a simple table
+name based on class name::
from sqlalchemy.ext.declarative import declared_attr
__mapper_args__ = {'polymorphic_identity': 'engineer'}
primary_language = Column(String(50))
-Alternatively, you can make the mixin intelligent enough to only
-return a ``__tablename__`` in the event that no table is already
-mapped in the inheritance hierarchy. To help with this, a
-:func:`~sqlalchemy.ext.declarative.has_inherited_table` helper
-function is provided that returns ``True`` if a parent class already
-has a mapped table.
-
-As an example, here's a mixin that will only allow single table
-inheritance::
+Alternatively, we can modify our ``__tablename__`` function to return
+``None`` for subclasses, using :func:`.has_inherited_table`. This has
+the effect of those subclasses being mapped with single table inheritance
+agaisnt the parent::
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.declarative import has_inherited_table
primary_language = Column(String(50))
__mapper_args__ = {'polymorphic_identity': 'engineer'}
+.. _mixin_inheritance_columns:
+
+Mixing in Columns in Inheritance Scenarios
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+In constrast to how ``__tablename__`` and other special names are handled when
+used with :class:`.declared_attr`, when we mix in columns and properties (e.g.
+relationships, column properties, etc.), the function is
+invoked for the **base class only** in the hierarchy. Below, only the
+``Person`` class will receive a column
+called ``id``; the mapping will fail on ``Engineer``, which is not given
+a primary key::
+
+ class HasId(object):
+ @declared_attr
+ def id(cls):
+ return Column('id', Integer, primary_key=True)
+
+ class Person(HasId, Base):
+ __tablename__ = 'person'
+ discriminator = Column('type', String(50))
+ __mapper_args__ = {'polymorphic_on': discriminator}
+
+ class Engineer(Person):
+ __tablename__ = 'engineer'
+ primary_language = Column(String(50))
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+
+It is usually the case in joined-table inheritance that we want distinctly
+named columns on each subclass. However in this case, we may want to have
+an ``id`` column on every table, and have them refer to each other via
+foreign key. We can achieve this as a mixin by using the
+:attr:`.declared_attr.cascading` modifier, which indicates that the
+function should be invoked **for each class in the hierarchy**, just like
+it does for ``__tablename__``::
+
+ class HasId(object):
+ @declared_attr.cascading
+ def id(cls):
+ if has_inherited_table(cls):
+ return Column('id',
+ Integer,
+ ForeignKey('person.id'), primary_key=True)
+ else:
+ return Column('id', Integer, primary_key=True)
+
+ class Person(HasId, Base):
+ __tablename__ = 'person'
+ discriminator = Column('type', String(50))
+ __mapper_args__ = {'polymorphic_on': discriminator}
+
+ class Engineer(Person):
+ __tablename__ = 'engineer'
+ primary_language = Column(String(50))
+ __mapper_args__ = {'polymorphic_identity': 'engineer'}
+
+
+.. versionadded:: 1.0.0 added :attr:`.declared_attr.cascading`.
Combining Table/Mapper Arguments from Multiple Mixins
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
from ...schema import Table, MetaData
-from ...orm import synonym as _orm_synonym, mapper,\
+from ...orm import synonym as _orm_synonym, \
comparable_property,\
- interfaces, properties
+ interfaces, properties, attributes
from ...orm.util import polymorphic_union
from ...orm.base import _mapper_or_none
-from ...util import OrderedDict
+from ...util import OrderedDict, hybridmethod, hybridproperty
+from ... import util
from ... import exc
import weakref
_declarative_constructor,\
_DeferredMapperConfig, _add_attribute
from .clsregistry import _class_resolver
-from . import clsregistry
def instrument_declarative(cls, registry, metadata):
"""
- def __init__(self, fget, *arg, **kw):
- super(declared_attr, self).__init__(fget, *arg, **kw)
+ def __init__(self, fget, cascading=False):
+ super(declared_attr, self).__init__(fget)
self.__doc__ = fget.__doc__
+ self._cascading = cascading
def __get__(desc, self, cls):
- return desc.fget(cls)
+ # use the ClassManager for memoization of values. This is better than
+ # adding yet another attribute onto the class, or using weakrefs
+ # here which are slow and take up memory. It also allows us to
+ # warn for non-mapped use of declared_attr.
+
+ manager = attributes.manager_of_class(cls)
+ if manager is None:
+ util.warn(
+ "Unmanaged access of declarative attribute %s from "
+ "non-mapped class %s" %
+ (desc.fget.__name__, cls.__name__))
+ return desc.fget(cls)
+ try:
+ reg = manager.info['declared_attr_reg']
+ except KeyError:
+ raise exc.InvalidRequestError(
+ "@declared_attr called outside of the "
+ "declarative mapping process; is declarative_base() being "
+ "used correctly?")
+
+ if desc in reg:
+ return reg[desc]
+ else:
+ reg[desc] = obj = desc.fget(cls)
+ return obj
+
+ @hybridmethod
+ def _stateful(cls, **kw):
+ return _stateful_declared_attr(**kw)
+
+ @hybridproperty
+ def cascading(cls):
+ """Mark a :class:`.declared_attr` as cascading.
+
+ This is a special-use modifier which indicates that a column
+ or MapperProperty-based declared attribute should be configured
+ distinctly per mapped subclass, within a mapped-inheritance scenario.
+
+ Below, both MyClass as well as MySubClass will have a distinct
+ ``id`` Column object established::
+
+ class HasSomeAttribute(object):
+ @declared_attr.cascading
+ def some_id(cls):
+ if has_inherited_table(cls):
+ return Column(
+ ForeignKey('myclass.id'), primary_key=True)
+ else:
+ return Column(Integer, primary_key=True)
+
+ return Column('id', Integer, primary_key=True)
+
+ class MyClass(HasSomeAttribute, Base):
+ ""
+ # ...
+
+ class MySubClass(MyClass):
+ ""
+ # ...
+
+ The behavior of the above configuration is that ``MySubClass``
+ will refer to both its own ``id`` column as well as that of
+ ``MyClass`` underneath the attribute named ``some_id``.
+
+ .. seealso::
+
+ :ref:`declarative_inheritance`
+
+ :ref:`mixin_inheritance_columns`
+
+
+ """
+ return cls._stateful(cascading=True)
+
+
+class _stateful_declared_attr(declared_attr):
+ def __init__(self, **kw):
+ self.kw = kw
+
+ def _stateful(self, **kw):
+ new_kw = self.kw.copy()
+ new_kw.update(kw)
+ return _stateful_declared_attr(**new_kw)
+
+ def __call__(self, fn):
+ return declared_attr(fn, **self.kw)
def declarative_base(bind=None, metadata=None, mapper=None, cls=object,
``__declare_last__()`` function, which is essentially
a hook for the :meth:`.after_configured` event.
- :class:`.AbstractConcreteBase` does not produce a mapped
- table for the class itself. Compare to :class:`.ConcreteBase`,
- which does.
+ :class:`.AbstractConcreteBase` does produce a mapped class
+ for the base class, however it is not persisted to any table; it
+ is instead mapped directly to the "polymorphic" selectable directly
+ and is only used for selecting. Compare to :class:`.ConcreteBase`,
+ which does create a persisted table for the base class.
Example::
employee_id = Column(Integer, primary_key=True)
name = Column(String(50))
manager_data = Column(String(40))
+
__mapper_args__ = {
- 'polymorphic_identity':'manager',
- 'concrete':True}
+ 'polymorphic_identity':'manager',
+ 'concrete':True}
+
+ The abstract base class is handled by declarative in a special way;
+ at class configuration time, it behaves like a declarative mixin
+ or an ``__abstract__`` base class. Once classes are configured
+ and mappings are produced, it then gets mapped itself, but
+ after all of its decscendants. This is a very unique system of mapping
+ not found in any other SQLAlchemy system.
+
+ Using this approach, we can specify columns and properties
+ that will take place on mapped subclasses, in the way that
+ we normally do as in :ref:`declarative_mixins`::
+
+ class Company(Base):
+ __tablename__ = 'company'
+ id = Column(Integer, primary_key=True)
+
+ class Employee(AbstractConcreteBase, Base):
+ employee_id = Column(Integer, primary_key=True)
+
+ @declared_attr
+ def company_id(cls):
+ return Column(ForeignKey('company.id'))
+
+ @declared_attr
+ def company(cls):
+ return relationship("Company")
+
+ class Manager(Employee):
+ __tablename__ = 'manager'
+
+ name = Column(String(50))
+ manager_data = Column(String(40))
+
+ __mapper_args__ = {
+ 'polymorphic_identity':'manager',
+ 'concrete':True}
+
+ When we make use of our mappings however, both ``Manager`` and
+ ``Employee`` will have an independently usable ``.company`` attribute::
+
+ session.query(Employee).filter(Employee.company.has(id=5))
+
+ .. versionchanged:: 1.0.0 - The mechanics of :class:`.AbstractConcreteBase`
+ have been reworked to support relationships established directly
+ on the abstract base, without any special configurational steps.
+
"""
- __abstract__ = True
+ __no_table__ = True
@classmethod
def __declare_first__(cls):
- if hasattr(cls, '__mapper__'):
+ cls._sa_decl_prepare_nocascade()
+
+ @classmethod
+ def _sa_decl_prepare_nocascade(cls):
+ if getattr(cls, '__mapper__', None):
return
- clsregistry.add_class(cls.__name__, cls)
+ to_map = _DeferredMapperConfig.config_for_cls(cls)
+
# can't rely on 'self_and_descendants' here
# since technically an immediate subclass
# might not be mapped, but a subclass
if mn is not None:
mappers.append(mn)
pjoin = cls._create_polymorphic_union(mappers)
- cls.__mapper__ = m = mapper(cls, pjoin, polymorphic_on=pjoin.c.type)
+
+ to_map.local_table = pjoin
+
+ m_args = to_map.mapper_args_fn or dict
+
+ def mapper_args():
+ args = m_args()
+ args['polymorphic_on'] = pjoin.c.type
+ return args
+ to_map.mapper_args_fn = mapper_args
+
+ m = to_map.map()
for scls in cls.__subclasses__():
sm = _mapper_or_none(scls)
from . import clsregistry
import collections
import weakref
+from sqlalchemy.orm import instrumentation
+
+declared_attr = declarative_props = None
def _declared_mapping_info(cls):
return None
+def _get_immediate_cls_attr(cls, attrname):
+ """return an attribute of the class that is either present directly
+ on the class, e.g. not on a superclass, or is from a superclass but
+ this superclass is a mixin, that is, not a descendant of
+ the declarative base.
+
+ This is used to detect attributes that indicate something about
+ a mapped class independently from any mapped classes that it may
+ inherit from.
+
+ """
+ for base in cls.__mro__:
+ _is_declarative_inherits = hasattr(base, '_decl_class_registry')
+ if attrname in base.__dict__:
+ value = getattr(base, attrname)
+ if (base is cls or
+ (base in cls.__bases__ and not _is_declarative_inherits)):
+ return value
+ else:
+ return None
+
+
def _as_declarative(cls, classname, dict_):
- from .api import declared_attr
+ global declared_attr, declarative_props
+ if declared_attr is None:
+ from .api import declared_attr
+ declarative_props = (declared_attr, util.classproperty)
- # dict_ will be a dictproxy, which we can't write to, and we need to!
- dict_ = dict(dict_)
+ if _get_immediate_cls_attr(cls, '__abstract__'):
+ return
- column_copies = {}
- potential_columns = {}
+ _MapperConfig.setup_mapping(cls, classname, dict_)
- mapper_args_fn = None
- table_args = inherited_table_args = None
- tablename = None
- declarative_props = (declared_attr, util.classproperty)
+class _MapperConfig(object):
- for base in cls.__mro__:
- _is_declarative_inherits = hasattr(base, '_decl_class_registry')
+ @classmethod
+ def setup_mapping(cls, cls_, classname, dict_):
+ defer_map = _get_immediate_cls_attr(
+ cls_, '_sa_decl_prepare_nocascade') or \
+ hasattr(cls_, '_sa_decl_prepare')
- if '__declare_last__' in base.__dict__:
- @event.listens_for(mapper, "after_configured")
- def go():
- cls.__declare_last__()
- if '__declare_first__' in base.__dict__:
- @event.listens_for(mapper, "before_configured")
- def go():
- cls.__declare_first__()
- if '__abstract__' in base.__dict__ and base.__abstract__:
- if (base is cls or
- (base in cls.__bases__ and not _is_declarative_inherits)):
- return
+ if defer_map:
+ cfg_cls = _DeferredMapperConfig
+ else:
+ cfg_cls = _MapperConfig
+ cfg_cls(cls_, classname, dict_)
- class_mapped = _declared_mapping_info(base) is not None
+ def __init__(self, cls_, classname, dict_):
- for name, obj in vars(base).items():
- if name == '__mapper_args__':
- if not mapper_args_fn and (
- not class_mapped or
- isinstance(obj, declarative_props)
- ):
- # don't even invoke __mapper_args__ until
- # after we've determined everything about the
- # mapped table.
- # make a copy of it so a class-level dictionary
- # is not overwritten when we update column-based
- # arguments.
- mapper_args_fn = lambda: dict(cls.__mapper_args__)
- elif name == '__tablename__':
- if not tablename and (
- not class_mapped or
- isinstance(obj, declarative_props)
- ):
- tablename = cls.__tablename__
- elif name == '__table_args__':
- if not table_args and (
- not class_mapped or
- isinstance(obj, declarative_props)
- ):
- table_args = cls.__table_args__
- if not isinstance(table_args, (tuple, dict, type(None))):
- raise exc.ArgumentError(
- "__table_args__ value must be a tuple, "
- "dict, or None")
- if base is not cls:
- inherited_table_args = True
- elif class_mapped:
- if isinstance(obj, declarative_props):
- util.warn("Regular (i.e. not __special__) "
- "attribute '%s.%s' uses @declared_attr, "
- "but owning class %s is mapped - "
- "not applying to subclass %s."
- % (base.__name__, name, base, cls))
- continue
- elif base is not cls:
- # we're a mixin.
- if isinstance(obj, Column):
- if getattr(cls, name) is not obj:
- # if column has been overridden
- # (like by the InstrumentedAttribute of the
- # superclass), skip
+ self.cls = cls_
+
+ # dict_ will be a dictproxy, which we can't write to, and we need to!
+ self.dict_ = dict(dict_)
+ self.classname = classname
+ self.mapped_table = None
+ self.properties = util.OrderedDict()
+ self.declared_columns = set()
+ self.column_copies = {}
+ self._setup_declared_events()
+
+ # register up front, so that @declared_attr can memoize
+ # function evaluations in .info
+ manager = instrumentation.register_class(self.cls)
+ manager.info['declared_attr_reg'] = {}
+
+ self._scan_attributes()
+
+ clsregistry.add_class(self.classname, self.cls)
+
+ self._extract_mappable_attributes()
+
+ self._extract_declared_columns()
+
+ self._setup_table()
+
+ self._setup_inheritance()
+
+ self._early_mapping()
+
+ def _early_mapping(self):
+ self.map()
+
+ def _setup_declared_events(self):
+ if _get_immediate_cls_attr(self.cls, '__declare_last__'):
+ @event.listens_for(mapper, "after_configured")
+ def after_configured():
+ self.cls.__declare_last__()
+
+ if _get_immediate_cls_attr(self.cls, '__declare_first__'):
+ @event.listens_for(mapper, "before_configured")
+ def before_configured():
+ self.cls.__declare_first__()
+
+ def _scan_attributes(self):
+ cls = self.cls
+ dict_ = self.dict_
+ column_copies = self.column_copies
+ mapper_args_fn = None
+ table_args = inherited_table_args = None
+ tablename = None
+
+ for base in cls.__mro__:
+ class_mapped = base is not cls and \
+ _declared_mapping_info(base) is not None and \
+ not _get_immediate_cls_attr(base, '_sa_decl_prepare_nocascade')
+
+ if not class_mapped and base is not cls:
+ self._produce_column_copies(base)
+
+ for name, obj in vars(base).items():
+ if name == '__mapper_args__':
+ if not mapper_args_fn and (
+ not class_mapped or
+ isinstance(obj, declarative_props)
+ ):
+ # don't even invoke __mapper_args__ until
+ # after we've determined everything about the
+ # mapped table.
+ # make a copy of it so a class-level dictionary
+ # is not overwritten when we update column-based
+ # arguments.
+ mapper_args_fn = lambda: dict(cls.__mapper_args__)
+ elif name == '__tablename__':
+ if not tablename and (
+ not class_mapped or
+ isinstance(obj, declarative_props)
+ ):
+ tablename = cls.__tablename__
+ elif name == '__table_args__':
+ if not table_args and (
+ not class_mapped or
+ isinstance(obj, declarative_props)
+ ):
+ table_args = cls.__table_args__
+ if not isinstance(
+ table_args, (tuple, dict, type(None))):
+ raise exc.ArgumentError(
+ "__table_args__ value must be a tuple, "
+ "dict, or None")
+ if base is not cls:
+ inherited_table_args = True
+ elif class_mapped:
+ if isinstance(obj, declarative_props):
+ util.warn("Regular (i.e. not __special__) "
+ "attribute '%s.%s' uses @declared_attr, "
+ "but owning class %s is mapped - "
+ "not applying to subclass %s."
+ % (base.__name__, name, base, cls))
+ continue
+ elif base is not cls:
+ # we're a mixin, abstract base, or something that is
+ # acting like that for now.
+ if isinstance(obj, Column):
+ # already copied columns to the mapped class.
continue
- if obj.foreign_keys:
+ elif isinstance(obj, MapperProperty):
raise exc.InvalidRequestError(
- "Columns with foreign keys to other columns "
- "must be declared as @declared_attr callables "
- "on declarative mixin classes. ")
- if name not in dict_ and not (
- '__table__' in dict_ and
- (obj.name or name) in dict_['__table__'].c
- ) and name not in potential_columns:
- potential_columns[name] = \
- column_copies[obj] = \
- obj.copy()
- column_copies[obj]._creation_order = \
- obj._creation_order
- elif isinstance(obj, MapperProperty):
+ "Mapper properties (i.e. deferred,"
+ "column_property(), relationship(), etc.) must "
+ "be declared as @declared_attr callables "
+ "on declarative mixin classes.")
+ elif isinstance(obj, declarative_props):
+ oldclassprop = isinstance(obj, util.classproperty)
+ if not oldclassprop and obj._cascading:
+ dict_[name] = column_copies[obj] = \
+ ret = obj.__get__(obj, cls)
+ else:
+ if oldclassprop:
+ util.warn_deprecated(
+ "Use of sqlalchemy.util.classproperty on "
+ "declarative classes is deprecated.")
+ dict_[name] = column_copies[obj] = \
+ ret = getattr(cls, name)
+ if isinstance(ret, (Column, MapperProperty)) and \
+ ret.doc is None:
+ ret.doc = obj.__doc__
+
+ if inherited_table_args and not tablename:
+ table_args = None
+
+ self.table_args = table_args
+ self.tablename = tablename
+ self.mapper_args_fn = mapper_args_fn
+
+ def _produce_column_copies(self, base):
+ cls = self.cls
+ dict_ = self.dict_
+ column_copies = self.column_copies
+ # copy mixin columns to the mapped class
+ for name, obj in vars(base).items():
+ if isinstance(obj, Column):
+ if getattr(cls, name) is not obj:
+ # if column has been overridden
+ # (like by the InstrumentedAttribute of the
+ # superclass), skip
+ continue
+ elif obj.foreign_keys:
raise exc.InvalidRequestError(
- "Mapper properties (i.e. deferred,"
- "column_property(), relationship(), etc.) must "
- "be declared as @declared_attr callables "
- "on declarative mixin classes.")
- elif isinstance(obj, declarative_props):
- dict_[name] = ret = \
- column_copies[obj] = getattr(cls, name)
- if isinstance(ret, (Column, MapperProperty)) and \
- ret.doc is None:
- ret.doc = obj.__doc__
-
- # apply inherited columns as we should
- for k, v in potential_columns.items():
- dict_[k] = v
-
- if inherited_table_args and not tablename:
- table_args = None
-
- clsregistry.add_class(classname, cls)
- our_stuff = util.OrderedDict()
-
- for k in list(dict_):
-
- # TODO: improve this ? all dunders ?
- if k in ('__table__', '__tablename__', '__mapper_args__'):
- continue
-
- value = dict_[k]
- if isinstance(value, declarative_props):
- value = getattr(cls, k)
-
- elif isinstance(value, QueryableAttribute) and \
- value.class_ is not cls and \
- value.key != k:
- # detect a QueryableAttribute that's already mapped being
- # assigned elsewhere in userland, turn into a synonym()
- value = synonym(value.key)
- setattr(cls, k, value)
-
- if (isinstance(value, tuple) and len(value) == 1 and
- isinstance(value[0], (Column, MapperProperty))):
- util.warn("Ignoring declarative-like tuple value of attribute "
- "%s: possibly a copy-and-paste error with a comma "
- "left at the end of the line?" % k)
- continue
- if not isinstance(value, (Column, MapperProperty)):
- if not k.startswith('__'):
- dict_.pop(k)
- setattr(cls, k, value)
- continue
- if k == 'metadata':
- raise exc.InvalidRequestError(
- "Attribute name 'metadata' is reserved "
- "for the MetaData instance when using a "
- "declarative base class."
- )
- prop = clsregistry._deferred_relationship(cls, value)
- our_stuff[k] = prop
-
- # set up attributes in the order they were created
- our_stuff.sort(key=lambda key: our_stuff[key]._creation_order)
-
- # extract columns from the class dict
- declared_columns = set()
- name_to_prop_key = collections.defaultdict(set)
- for key, c in list(our_stuff.items()):
- if isinstance(c, (ColumnProperty, CompositeProperty)):
- for col in c.columns:
- if isinstance(col, Column) and \
- col.table is None:
- _undefer_column_name(key, col)
- if not isinstance(c, CompositeProperty):
- name_to_prop_key[col.name].add(key)
- declared_columns.add(col)
- elif isinstance(c, Column):
- _undefer_column_name(key, c)
- name_to_prop_key[c.name].add(key)
- declared_columns.add(c)
- # if the column is the same name as the key,
- # remove it from the explicit properties dict.
- # the normal rules for assigning column-based properties
- # will take over, including precedence of columns
- # in multi-column ColumnProperties.
- if key == c.key:
- del our_stuff[key]
-
- for name, keys in name_to_prop_key.items():
- if len(keys) > 1:
- util.warn(
- "On class %r, Column object %r named directly multiple times, "
- "only one will be used: %s" %
- (classname, name, (", ".join(sorted(keys))))
- )
+ "Columns with foreign keys to other columns "
+ "must be declared as @declared_attr callables "
+ "on declarative mixin classes. ")
+ elif name not in dict_ and not (
+ '__table__' in dict_ and
+ (obj.name or name) in dict_['__table__'].c
+ ):
+ column_copies[obj] = copy_ = obj.copy()
+ copy_._creation_order = obj._creation_order
+ setattr(cls, name, copy_)
+ dict_[name] = copy_
- declared_columns = sorted(
- declared_columns, key=lambda c: c._creation_order)
- table = None
+ def _extract_mappable_attributes(self):
+ cls = self.cls
+ dict_ = self.dict_
- if hasattr(cls, '__table_cls__'):
- table_cls = util.unbound_method_to_callable(cls.__table_cls__)
- else:
- table_cls = Table
-
- if '__table__' not in dict_:
- if tablename is not None:
-
- args, table_kw = (), {}
- if table_args:
- if isinstance(table_args, dict):
- table_kw = table_args
- elif isinstance(table_args, tuple):
- if isinstance(table_args[-1], dict):
- args, table_kw = table_args[0:-1], table_args[-1]
- else:
- args = table_args
-
- autoload = dict_.get('__autoload__')
- if autoload:
- table_kw['autoload'] = True
-
- cls.__table__ = table = table_cls(
- tablename, cls.metadata,
- *(tuple(declared_columns) + tuple(args)),
- **table_kw)
- else:
- table = cls.__table__
- if declared_columns:
- for c in declared_columns:
- if not table.c.contains_column(c):
- raise exc.ArgumentError(
- "Can't add additional column %r when "
- "specifying __table__" % c.key
- )
+ our_stuff = self.properties
- if hasattr(cls, '__mapper_cls__'):
- mapper_cls = util.unbound_method_to_callable(cls.__mapper_cls__)
- else:
- mapper_cls = mapper
+ for k in list(dict_):
- for c in cls.__bases__:
- if _declared_mapping_info(c) is not None:
- inherits = c
- break
- else:
- inherits = None
+ # TODO: improve this ? all dunders ?
+ if k in ('__table__', '__tablename__', '__mapper_args__'):
+ continue
- if table is None and inherits is None:
- raise exc.InvalidRequestError(
- "Class %r does not have a __table__ or __tablename__ "
- "specified and does not inherit from an existing "
- "table-mapped class." % cls
- )
- elif inherits:
- inherited_mapper = _declared_mapping_info(inherits)
- inherited_table = inherited_mapper.local_table
- inherited_mapped_table = inherited_mapper.mapped_table
-
- if table is None:
- # single table inheritance.
- # ensure no table args
- if table_args:
- raise exc.ArgumentError(
- "Can't place __table_args__ on an inherited class "
- "with no table."
+ value = dict_[k]
+ if isinstance(value, declarative_props):
+ value = getattr(cls, k)
+
+ elif isinstance(value, QueryableAttribute) and \
+ value.class_ is not cls and \
+ value.key != k:
+ # detect a QueryableAttribute that's already mapped being
+ # assigned elsewhere in userland, turn into a synonym()
+ value = synonym(value.key)
+ setattr(cls, k, value)
+
+ if (isinstance(value, tuple) and len(value) == 1 and
+ isinstance(value[0], (Column, MapperProperty))):
+ util.warn("Ignoring declarative-like tuple value of attribute "
+ "%s: possibly a copy-and-paste error with a comma "
+ "left at the end of the line?" % k)
+ continue
+ if not isinstance(value, (Column, MapperProperty)):
+ if not k.startswith('__'):
+ dict_.pop(k)
+ setattr(cls, k, value)
+ continue
+ if k == 'metadata':
+ raise exc.InvalidRequestError(
+ "Attribute name 'metadata' is reserved "
+ "for the MetaData instance when using a "
+ "declarative base class."
+ )
+ prop = clsregistry._deferred_relationship(cls, value)
+ our_stuff[k] = prop
+
+ def _extract_declared_columns(self):
+ our_stuff = self.properties
+
+ # set up attributes in the order they were created
+ our_stuff.sort(key=lambda key: our_stuff[key]._creation_order)
+
+ # extract columns from the class dict
+ declared_columns = self.declared_columns
+ name_to_prop_key = collections.defaultdict(set)
+ for key, c in list(our_stuff.items()):
+ if isinstance(c, (ColumnProperty, CompositeProperty)):
+ for col in c.columns:
+ if isinstance(col, Column) and \
+ col.table is None:
+ _undefer_column_name(key, col)
+ if not isinstance(c, CompositeProperty):
+ name_to_prop_key[col.name].add(key)
+ declared_columns.add(col)
+ elif isinstance(c, Column):
+ _undefer_column_name(key, c)
+ name_to_prop_key[c.name].add(key)
+ declared_columns.add(c)
+ # if the column is the same name as the key,
+ # remove it from the explicit properties dict.
+ # the normal rules for assigning column-based properties
+ # will take over, including precedence of columns
+ # in multi-column ColumnProperties.
+ if key == c.key:
+ del our_stuff[key]
+
+ for name, keys in name_to_prop_key.items():
+ if len(keys) > 1:
+ util.warn(
+ "On class %r, Column object %r named "
+ "directly multiple times, "
+ "only one will be used: %s" %
+ (self.classname, name, (", ".join(sorted(keys))))
)
- # add any columns declared here to the inherited table.
- for c in declared_columns:
- if c.primary_key:
- raise exc.ArgumentError(
- "Can't place primary key columns on an inherited "
- "class with no table."
- )
- if c.name in inherited_table.c:
- if inherited_table.c[c.name] is c:
- continue
- raise exc.ArgumentError(
- "Column '%s' on class %s conflicts with "
- "existing column '%s'" %
- (c, cls, inherited_table.c[c.name])
- )
- inherited_table.append_column(c)
- if inherited_mapped_table is not None and \
- inherited_mapped_table is not inherited_table:
- inherited_mapped_table._refresh_for_new_column(c)
-
- defer_map = hasattr(cls, '_sa_decl_prepare')
- if defer_map:
- cfg_cls = _DeferredMapperConfig
- else:
- cfg_cls = _MapperConfig
- mt = cfg_cls(mapper_cls,
- cls, table,
- inherits,
- declared_columns,
- column_copies,
- our_stuff,
- mapper_args_fn)
- if not defer_map:
- mt.map()
+ def _setup_table(self):
+ cls = self.cls
+ tablename = self.tablename
+ table_args = self.table_args
+ dict_ = self.dict_
+ declared_columns = self.declared_columns
-class _MapperConfig(object):
+ declared_columns = self.declared_columns = sorted(
+ declared_columns, key=lambda c: c._creation_order)
+ table = None
- mapped_table = None
-
- def __init__(self, mapper_cls,
- cls,
- table,
- inherits,
- declared_columns,
- column_copies,
- properties, mapper_args_fn):
- self.mapper_cls = mapper_cls
- self.cls = cls
+ if hasattr(cls, '__table_cls__'):
+ table_cls = util.unbound_method_to_callable(cls.__table_cls__)
+ else:
+ table_cls = Table
+
+ if '__table__' not in dict_:
+ if tablename is not None:
+
+ args, table_kw = (), {}
+ if table_args:
+ if isinstance(table_args, dict):
+ table_kw = table_args
+ elif isinstance(table_args, tuple):
+ if isinstance(table_args[-1], dict):
+ args, table_kw = table_args[0:-1], table_args[-1]
+ else:
+ args = table_args
+
+ autoload = dict_.get('__autoload__')
+ if autoload:
+ table_kw['autoload'] = True
+
+ cls.__table__ = table = table_cls(
+ tablename, cls.metadata,
+ *(tuple(declared_columns) + tuple(args)),
+ **table_kw)
+ else:
+ table = cls.__table__
+ if declared_columns:
+ for c in declared_columns:
+ if not table.c.contains_column(c):
+ raise exc.ArgumentError(
+ "Can't add additional column %r when "
+ "specifying __table__" % c.key
+ )
self.local_table = table
- self.inherits = inherits
- self.properties = properties
- self.mapper_args_fn = mapper_args_fn
- self.declared_columns = declared_columns
- self.column_copies = column_copies
+
+ def _setup_inheritance(self):
+ table = self.local_table
+ cls = self.cls
+ table_args = self.table_args
+ declared_columns = self.declared_columns
+ for c in cls.__bases__:
+ if _declared_mapping_info(c) is not None and \
+ not _get_immediate_cls_attr(
+ c, '_sa_decl_prepare_nocascade'):
+ self.inherits = c
+ break
+ else:
+ self.inherits = None
+
+ if table is None and self.inherits is None and \
+ not _get_immediate_cls_attr(cls, '__no_table__'):
+
+ raise exc.InvalidRequestError(
+ "Class %r does not have a __table__ or __tablename__ "
+ "specified and does not inherit from an existing "
+ "table-mapped class." % cls
+ )
+ elif self.inherits:
+ inherited_mapper = _declared_mapping_info(self.inherits)
+ inherited_table = inherited_mapper.local_table
+ inherited_mapped_table = inherited_mapper.mapped_table
+
+ if table is None:
+ # single table inheritance.
+ # ensure no table args
+ if table_args:
+ raise exc.ArgumentError(
+ "Can't place __table_args__ on an inherited class "
+ "with no table."
+ )
+ # add any columns declared here to the inherited table.
+ for c in declared_columns:
+ if c.primary_key:
+ raise exc.ArgumentError(
+ "Can't place primary key columns on an inherited "
+ "class with no table."
+ )
+ if c.name in inherited_table.c:
+ if inherited_table.c[c.name] is c:
+ continue
+ raise exc.ArgumentError(
+ "Column '%s' on class %s conflicts with "
+ "existing column '%s'" %
+ (c, cls, inherited_table.c[c.name])
+ )
+ inherited_table.append_column(c)
+ if inherited_mapped_table is not None and \
+ inherited_mapped_table is not inherited_table:
+ inherited_mapped_table._refresh_for_new_column(c)
def _prepare_mapper_arguments(self):
properties = self.properties
properties[k] = [col] + p.columns
result_mapper_args = mapper_args.copy()
result_mapper_args['properties'] = properties
- return result_mapper_args
+ self.mapper_args = result_mapper_args
def map(self):
- mapper_args = self._prepare_mapper_arguments()
- self.cls.__mapper__ = self.mapper_cls(
+ self._prepare_mapper_arguments()
+ if hasattr(self.cls, '__mapper_cls__'):
+ mapper_cls = util.unbound_method_to_callable(
+ self.cls.__mapper_cls__)
+ else:
+ mapper_cls = mapper
+
+ self.cls.__mapper__ = mp_ = mapper_cls(
self.cls,
self.local_table,
- **mapper_args
+ **self.mapper_args
)
+ del mp_.class_manager.info['declared_attr_reg']
+ return mp_
class _DeferredMapperConfig(_MapperConfig):
_configs = util.OrderedDict()
+ def _early_mapping(self):
+ pass
+
@property
def cls(self):
return self._cls()
def map(self):
self._configs.pop(self._cls, None)
- super(_DeferredMapperConfig, self).map()
+ return super(_DeferredMapperConfig, self).map()
def _add_attribute(cls, key, value):
auto-session attachment logic.
"""
+
+ # when using declarative as of 1.0, the register_class has
+ # already happened from within declarative.
manager = attributes.manager_of_class(self.class_)
if self.non_primary:
"create a non primary Mapper. clear_mappers() will "
"remove *all* current mappers from all classes." %
self.class_)
- # else:
- # a ClassManager may already exist as
- # ClassManager.instrument_attribute() creates
- # new managers for each subclass if they don't yet exist.
+
+ if manager is None:
+ manager = instrumentation.register_class(self.class_)
_mapper_registry[self] = True
self.dispatch.instrument_class(self, self.class_)
- if manager is None:
- manager = instrumentation.register_class(self.class_)
-
self.class_manager = manager
manager.mapper = self
existing = getattr(self, 'table', None)
if existing is not None and existing is not table:
raise exc.ArgumentError(
- "Column object already assigned to Table '%s'" %
- existing.description)
+ "Column object '%s' already assigned to Table '%s'" % (
+ self.key,
+ existing.description
+ ))
if self.key in table._columns:
col = table._columns.get(self.key)
duck_type_collection, assert_arg_type, symbol, dictlike_iteritems,\
classproperty, set_creation_order, warn_exception, warn, NoneType,\
constructor_copy, methods_equivalent, chop_traceback, asint,\
- generic_repr, counter, PluginLoader, hybridmethod, safe_reraise,\
+ generic_repr, counter, PluginLoader, hybridproperty, hybridmethod, \
+ safe_reraise,\
get_callable_argspec, only_once, attrsetter, ellipses_string, \
warn_limited
return desc.fget(cls)
+class hybridproperty(object):
+ def __init__(self, func):
+ self.func = func
+
+ def __get__(self, instance, owner):
+ if instance is None:
+ clsval = self.func(owner)
+ clsval.__doc__ = self.func.__doc__
+ return clsval
+ else:
+ return self.func(instance)
+
+
class hybridmethod(object):
"""Decorate a function as cls- or instance- level."""
- def __init__(self, func, expr=None):
+ def __init__(self, func):
self.func = func
def __get__(self, instance, owner):
polymorphic_union, deferred, Session
from sqlalchemy.ext.declarative import declared_attr, AbstractConcreteBase, \
ConcreteBase, has_inherited_table
-from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import fixtures, mock
Base = None
"b.b_data AS b_data, 'b' AS type FROM b) AS pjoin "
"ON pjoin.a_id = a.id"
)
+
+ def test_prop_on_base(self):
+ """test [ticket:2670] """
+
+ counter = mock.Mock()
+
+ class Something(Base):
+ __tablename__ = 'something'
+ id = Column(Integer, primary_key=True)
+
+ class AbstractConcreteAbstraction(AbstractConcreteBase, Base):
+ id = Column(Integer, primary_key=True)
+ x = Column(Integer)
+ y = Column(Integer)
+
+ @declared_attr
+ def something_id(cls):
+ return Column(ForeignKey(Something.id))
+
+ @declared_attr
+ def something(cls):
+ counter(cls, "something")
+ return relationship("Something")
+
+ @declared_attr
+ def something_else(cls):
+ counter(cls, "something_else")
+ return relationship("Something")
+
+ class ConcreteConcreteAbstraction(AbstractConcreteAbstraction):
+ __tablename__ = 'cca'
+ __mapper_args__ = {
+ 'polymorphic_identity': 'ccb',
+ 'concrete': True}
+
+ # concrete is mapped, the abstract base is not (yet)
+ assert ConcreteConcreteAbstraction.__mapper__
+ assert not hasattr(AbstractConcreteAbstraction, '__mapper__')
+
+ session = Session()
+ self.assert_compile(
+ session.query(ConcreteConcreteAbstraction).filter(
+ ConcreteConcreteAbstraction.something.has(id=1)),
+ "SELECT cca.id AS cca_id, cca.x AS cca_x, cca.y AS cca_y, "
+ "cca.something_id AS cca_something_id FROM cca WHERE EXISTS "
+ "(SELECT 1 FROM something WHERE something.id = cca.something_id "
+ "AND something.id = :id_1)"
+ )
+
+ # now it is
+ assert AbstractConcreteAbstraction.__mapper__
+
+ self.assert_compile(
+ session.query(ConcreteConcreteAbstraction).filter(
+ ConcreteConcreteAbstraction.something_else.has(id=1)),
+ "SELECT cca.id AS cca_id, cca.x AS cca_x, cca.y AS cca_y, "
+ "cca.something_id AS cca_something_id FROM cca WHERE EXISTS "
+ "(SELECT 1 FROM something WHERE something.id = cca.something_id "
+ "AND something.id = :id_1)"
+ )
+
+ self.assert_compile(
+ session.query(AbstractConcreteAbstraction).filter(
+ AbstractConcreteAbstraction.something.has(id=1)),
+ "SELECT pjoin.id AS pjoin_id, pjoin.x AS pjoin_x, "
+ "pjoin.y AS pjoin_y, pjoin.something_id AS pjoin_something_id, "
+ "pjoin.type AS pjoin_type FROM "
+ "(SELECT cca.id AS id, cca.x AS x, cca.y AS y, "
+ "cca.something_id AS something_id, 'ccb' AS type FROM cca) "
+ "AS pjoin WHERE EXISTS (SELECT 1 FROM something "
+ "WHERE something.id = pjoin.something_id AND something.id = :id_1)"
+ )
+
+ self.assert_compile(
+ session.query(AbstractConcreteAbstraction).filter(
+ AbstractConcreteAbstraction.something_else.has(id=1)),
+ "SELECT pjoin.id AS pjoin_id, pjoin.x AS pjoin_x, "
+ "pjoin.y AS pjoin_y, pjoin.something_id AS pjoin_something_id, "
+ "pjoin.type AS pjoin_type FROM "
+ "(SELECT cca.id AS id, cca.x AS x, cca.y AS y, "
+ "cca.something_id AS something_id, 'ccb' AS type FROM cca) "
+ "AS pjoin WHERE EXISTS (SELECT 1 FROM something "
+ "WHERE something.id = pjoin.something_id AND something.id = :id_1)"
+ )
+
from sqlalchemy.ext import declarative as decl
import sqlalchemy as sa
from sqlalchemy import testing
-from sqlalchemy import Integer, String, ForeignKey
+from sqlalchemy import Integer, String, ForeignKey, select, func
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import relationship, create_session, class_mapper, \
configure_mappers, clear_mappers, \
- deferred, column_property, \
- Session
+ deferred, column_property, Session, base as orm_base
from sqlalchemy.util import classproperty
from sqlalchemy.ext.declarative import declared_attr
-from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import fixtures, mock
+from sqlalchemy.testing.util import gc_collect
Base = None
self._test_relationship(True)
+class DeclaredAttrTest(DeclarativeTestBase, testing.AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_singleton_behavior_within_decl(self):
+ counter = mock.Mock()
+
+ class Mixin(object):
+ @declared_attr
+ def my_prop(cls):
+ counter(cls)
+ return Column('x', Integer)
+
+ class A(Base, Mixin):
+ __tablename__ = 'a'
+ id = Column(Integer, primary_key=True)
+
+ @declared_attr
+ def my_other_prop(cls):
+ return column_property(cls.my_prop + 5)
+
+ eq_(counter.mock_calls, [mock.call(A)])
+
+ class B(Base, Mixin):
+ __tablename__ = 'b'
+ id = Column(Integer, primary_key=True)
+
+ @declared_attr
+ def my_other_prop(cls):
+ return column_property(cls.my_prop + 5)
+
+ eq_(
+ counter.mock_calls,
+ [mock.call(A), mock.call(B)])
+
+ # this is why we need singleton-per-class behavior. We get
+ # an un-bound "x" column otherwise here, because my_prop() generates
+ # multiple columns.
+ a_col = A.my_other_prop.__clause_element__().element.left
+ b_col = B.my_other_prop.__clause_element__().element.left
+ is_(a_col.table, A.__table__)
+ is_(b_col.table, B.__table__)
+ is_(a_col, A.__table__.c.x)
+ is_(b_col, B.__table__.c.x)
+
+ s = Session()
+ self.assert_compile(
+ s.query(A),
+ "SELECT a.x AS a_x, a.x + :x_1 AS anon_1, a.id AS a_id FROM a"
+ )
+ self.assert_compile(
+ s.query(B),
+ "SELECT b.x AS b_x, b.x + :x_1 AS anon_1, b.id AS b_id FROM b"
+ )
+
+
+ def test_singleton_gc(self):
+ counter = mock.Mock()
+
+ class Mixin(object):
+ @declared_attr
+ def my_prop(cls):
+ counter(cls.__name__)
+ return Column('x', Integer)
+
+ class A(Base, Mixin):
+ __tablename__ = 'b'
+ id = Column(Integer, primary_key=True)
+
+ @declared_attr
+ def my_other_prop(cls):
+ return column_property(cls.my_prop + 5)
+
+ eq_(counter.mock_calls, [mock.call("A")])
+ del A
+ gc_collect()
+ assert "A" not in Base._decl_class_registry
+
+ def test_can_we_access_the_mixin_straight(self):
+ class Mixin(object):
+ @declared_attr
+ def my_prop(cls):
+ return Column('x', Integer)
+
+ assert_raises_message(
+ sa.exc.SAWarning,
+ "Unmanaged access of declarative attribute my_prop "
+ "from non-mapped class Mixin",
+ getattr, Mixin, "my_prop"
+ )
+
+ def test_property_noncascade(self):
+ counter = mock.Mock()
+
+ class Mixin(object):
+ @declared_attr
+ def my_prop(cls):
+ counter(cls)
+ return column_property(cls.x + 2)
+
+ class A(Base, Mixin):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+ x = Column(Integer)
+
+ class B(A):
+ pass
+
+ eq_(counter.mock_calls, [mock.call(A)])
+
+ def test_property_cascade(self):
+ counter = mock.Mock()
+
+ class Mixin(object):
+ @declared_attr.cascading
+ def my_prop(cls):
+ counter(cls)
+ return column_property(cls.x + 2)
+
+ class A(Base, Mixin):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+ x = Column(Integer)
+
+ class B(A):
+ pass
+
+ eq_(counter.mock_calls, [mock.call(A), mock.call(B)])
+
+ def test_column_pre_map(self):
+ counter = mock.Mock()
+
+ class Mixin(object):
+ @declared_attr
+ def my_col(cls):
+ counter(cls)
+ assert not orm_base._mapper_or_none(cls)
+ return Column('x', Integer)
+
+ class A(Base, Mixin):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+
+ eq_(counter.mock_calls, [mock.call(A)])
+
+ def test_mixin_attr_refers_to_column_copies(self):
+ # this @declared_attr can refer to User.id
+ # freely because we now do the "copy column" operation
+ # before the declared_attr is invoked.
+
+ counter = mock.Mock()
+
+ class HasAddressCount(object):
+ id = Column(Integer, primary_key=True)
+
+ @declared_attr
+ def address_count(cls):
+ counter(cls.id)
+ return column_property(
+ select([func.count(Address.id)]).
+ where(Address.user_id == cls.id).
+ as_scalar()
+ )
+
+ class Address(Base):
+ __tablename__ = 'address'
+ id = Column(Integer, primary_key=True)
+ user_id = Column(ForeignKey('user.id'))
+
+ class User(Base, HasAddressCount):
+ __tablename__ = 'user'
+
+ eq_(
+ counter.mock_calls,
+ [mock.call(User.id)]
+ )
+
+ sess = Session()
+ self.assert_compile(
+ sess.query(User).having(User.address_count > 5),
+ 'SELECT (SELECT count(address.id) AS '
+ 'count_1 FROM address WHERE address.user_id = "user".id) '
+ 'AS anon_1, "user".id AS user_id FROM "user" '
+ 'HAVING (SELECT count(address.id) AS '
+ 'count_1 FROM address WHERE address.user_id = "user".id) '
+ '> :param_1'
+ )
+
+
class AbstractTest(DeclarativeTestBase):
def test_abstract_boolean(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
