- changed "instrument" argument on synonym() to "descriptor", for consistency with comparable_proeprty()
- extensions
- The "synonym" function is now directly usable with
"declarative". Pass in the decorated property using the
- "instrument" keyword argument, e.g.: somekey =
- synonym('_somekey', instrument=property(g, s))
-
+ "descriptor" keyword argument, e.g.: somekey =
+ synonym('_somekey', descriptor=property(g, s))
+
+ - the "deferred" function is usable with "declarative".
+ Simplest usage is to declare deferred and Column together,
+ e.g.: data = deferred(Column(Text))
+
- Declarative also gained @synonym_for(...) and
@comparable_using(...), front-ends for synonym and
comparable_property.
Synonyms are one area where ``declarative`` needs to slightly change the
usual SQLAlchemy configurational syntax. To define a getter/setter which
-proxies to an underlying attribute, use ``synonym`` with the ``instruments``
+proxies to an underlying attribute, use ``synonym`` with the ``descriptor``
argument::
class MyClass(Base):
return self._some_attr
def _set_attr(self, attr)
self._some_attr = attr
- attr = synonym('_attr', instruments=property(_get_attr, _set_attr))
+ attr = synonym('_attr', descriptor=property(_get_attr, _set_attr))
The above synonym is then usable as an instance attribute as well as a
class-level expression construct::
from sqlalchemy.schema import Table, SchemaItem, Column, MetaData
from sqlalchemy.orm import synonym as _orm_synonym, mapper, comparable_property
from sqlalchemy.orm.interfaces import MapperProperty
-from sqlalchemy.orm.properties import PropertyLoader
+from sqlalchemy.orm.properties import PropertyLoader, ColumnProperty
from sqlalchemy import util
__all__ = ['declarative_base', 'synonym_for', 'comparable_using',
table_kw = {}
cols = []
for key, c in our_stuff.iteritems():
- if isinstance(c, Column):
+ if isinstance(c, ColumnProperty):
+ for col in c.columns:
+ if isinstance(col, Column) and col.table is None:
+ _undefer_column_name(key, col)
+ cols.append(col)
+ elif isinstance(c, Column):
_undefer_column_name(key, c)
cols.append(c)
cls.__table__ = table = Table(tablename, cls.metadata,
return prop
def declared_synonym(prop, name):
- """deprecated. use synonym(name, instrument=prop)."""
+ """deprecated. use synonym(name, descriptor=prop)."""
- return _orm_synonym(name, instrument=prop)
+ return _orm_synonym(name, descriptor=prop)
declared_synonym = util.deprecated(declared_synonym)
def synonym_for(name, map_column=False):
"""Decorator, make a Python @property a query synonym for a column.
A decorator version of [sqlalchemy.orm#synonym()]. The function being
- decoratred is the 'instrument', otherwise passes its arguments through
+ decoratred is the 'descriptor', otherwise passes its arguments through
to synonym()::
@synonym_for('col')
The regular ``synonym()`` is also usable directly in a declarative
setting and may be convenient for read/write properties::
- prop = synonym('col', instrument=property(_read_prop, _write_prop))
+ prop = synonym('col', descriptor=property(_read_prop, _write_prop))
"""
def decorate(fn):
- return _orm_synonym(name, map_column=map_column, instrument=fn)
+ return _orm_synonym(name, map_column=map_column, descriptor=fn)
return decorate
return Mapper(class_, local_table, *args, **params)
-def synonym(name, map_column=False, instrument=None, proxy=False):
+def synonym(name, map_column=False, descriptor=None, proxy=False):
"""Set up `name` as a synonym to another mapped property.
Used with the ``properties`` dictionary sent to [sqlalchemy.orm#mapper()].
is not already available.
"""
- return SynonymProperty(name, map_column=map_column, instrument=instrument)
+ return SynonymProperty(name, map_column=map_column, descriptor=descriptor)
def comparable_property(comparator_factory, descriptor=None):
"""Provide query semantics for an unmanaged attribute.
elif isinstance(prop, SynonymProperty) and setparent:
- if prop.instrument is None:
- prop.instrument = getattr(self.class_, key, None)
- if isinstance(prop.instrument, Mapper._CompileOnAttr):
- prop.instrument = object.__getattribute__(prop.instrument, 'existing_prop')
+ if prop.descriptor is None:
+ prop.descriptor = getattr(self.class_, key, None)
+ if isinstance(prop.descriptor, Mapper._CompileOnAttr):
+ prop.descriptor = object.__getattribute__(prop.descriptor, 'existing_prop')
if prop.map_column:
if not key in self.mapped_table.c:
raise exceptions.ArgumentError("Can't compile synonym '%s': no column on table '%s' named '%s'" % (prop.name, self.mapped_table.description, key))
other.__composite_values__())])
class SynonymProperty(MapperProperty):
- def __init__(self, name, map_column=None, instrument=None):
+ def __init__(self, name, map_column=None, descriptor=None):
self.name = name
self.map_column=map_column
- self.instrument = instrument
+ self.descriptor = descriptor
def setup(self, querycontext, **kwargs):
pass
def comparator():
return self.parent._get_property(self.key, resolve_synonyms=True).comparator
self.logger.info("register managed attribute %s on class %s" % (self.key, class_.__name__))
- if self.instrument is None:
+ if self.descriptor is None:
class SynonymProp(object):
def __set__(s, obj, value):
setattr(obj, self.name, value)
if obj is None:
return s
return getattr(obj, self.name)
- self.instrument = SynonymProp()
- sessionlib.register_attribute(class_, self.key, uselist=False, proxy_property=self.instrument, useobject=False, comparator=comparator)
+ self.descriptor = SynonymProp()
+ sessionlib.register_attribute(class_, self.key, uselist=False, proxy_property=self.descriptor, useobject=False, comparator=comparator)
def merge(self, session, source, dest, _recursive):
pass
from testlib import *
-class DeclarativeTest(TestBase):
+class DeclarativeTest(TestBase, AssertsExecutionResults):
def setUp(self):
global Base
Base = declarative_base(testing.db)
self.assertEquals(sess.query(User).all(),
[User(name='u1', a='a', b='b')])
+ def test_column_properties(self):
+
+ class Address(Base, Fixture):
+ __tablename__ = 'addresses'
+ id = Column(Integer, primary_key=True)
+ email = Column(String(50))
+ user_id = Column(Integer, ForeignKey('users.id'))
+
+ class User(Base, Fixture):
+ __tablename__ = 'users'
+
+ id = Column('id', Integer, primary_key=True)
+ name = Column('name', String(50))
+ adr_count = column_property(select([func.count(Address.id)], Address.user_id==id).as_scalar())
+ addresses = relation(Address)
+
+ Base.metadata.create_all()
+
+ u1 = User(name='u1', addresses=[
+ Address(email='one'),
+ Address(email='two'),
+ ])
+ sess = create_session()
+ sess.save(u1)
+ sess.flush()
+ sess.clear()
+
+ self.assertEquals(sess.query(User).all(), [User(name='u1', adr_count=2, addresses=[
+ Address(email='one'),
+ Address(email='two'),
+ ])])
+
+ def test_column_properties_2(self):
+
+ class Address(Base, Fixture):
+ __tablename__ = 'addresses'
+ id = Column(Integer, primary_key=True)
+ email = Column(String(50))
+ user_id = Column(Integer, ForeignKey('users.id'))
+
+ class User(Base, Fixture):
+ __tablename__ = 'users'
+
+ id = Column('id', Integer, primary_key=True)
+ name = Column('name', String(50))
+ # this is not "valid" but we want to test that Address.id doesnt get stuck into user's table
+ adr_count = Address.id
+
+ self.assertEquals(set(User.__table__.c.keys()), set(['id', 'name']))
+ self.assertEquals(set(Address.__table__.c.keys()), set(['id', 'email', 'user_id']))
+
+ def test_deferred(self):
+ class User(Base, Fixture):
+ __tablename__ = 'users'
+
+ id = Column(Integer, primary_key=True)
+ name = deferred(Column(String(50)))
+
+ Base.metadata.create_all()
+ sess = create_session()
+ sess.save(User(name='u1'))
+ sess.flush()
+ sess.clear()
+
+ u1 = sess.query(User).filter(User.name=='u1').one()
+ assert 'name' not in u1.__dict__
+ def go():
+ assert u1.name == 'u1'
+ self.assert_sql_count(testing.db, go, 1)
+
def test_synonym_inline(self):
class User(Base, Fixture):
__tablename__ = 'users'
self._name = "SOMENAME " + name
def _get_name(self):
return self._name
- name = synonym('_name', instrument=property(_get_name, _set_name))
+ name = synonym('_name', descriptor=property(_get_name, _set_name))
Base.metadata.create_all()
def _get_name(self):
return self._name
name = property(_get_name, _set_name)
- User.name = synonym('_name', instrument=User.name)
+ User.name = synonym('_name', descriptor=User.name)
Base.metadata.create_all()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
