from sqlalchemy.ext.declarative import declarative_base
from history_meta import VersionedMeta, VersionedListener
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
-from sqlalchemy.orm import clear_mappers, compile_mappers, sessionmaker, deferred
+from sqlalchemy.orm import clear_mappers, sessionmaker, deferred
from sqlalchemy.test.testing import TestBase, eq_
from sqlalchemy.test.entities import ComparableEntity
Mapper,
_mapper_registry,
class_mapper,
+ configure_mappers
)
from sqlalchemy.orm.interfaces import (
EXT_CONTINUE,
'column_property',
'comparable_property',
'compile_mappers',
+ 'configure_mappers',
'composite',
'contains_alias',
'contains_eager',
"""
return ComparableProperty(comparator_factory, descriptor)
-
+
+@sa_util.deprecated("0.7", message=":func:`.compile_mappers` "
+ "is renamed to :func:`.configure_mappers`")
def compile_mappers():
- """Compile all mappers that have been defined.
-
- This is equivalent to calling ``compile()`` on any individual mapper.
-
- """
- for m in list(_mapper_registry):
- m.compile()
+ """Initialize the inter-mapper relationships of all mappers that have been defined."""
+
+ configure_mappers()
def clear_mappers():
"""Remove all mappers from all classes.
_already_compiling = False
_none_set = frozenset([None])
-_memoized_compiled_property = util.group_expirable_memoized_property()
+_memoized_configured_property = util.group_expirable_memoized_property()
# a list of MapperExtensions that will be installed in all mappers by default
global_extensions = []
else:
self.exclude_properties = None
- self.compiled = False
+ self.configured = False
# prevent this mapper from being constructed
- # while a compile() is occuring (and defer a compile()
+ # while a configure_mappers() is occuring (and defer a configure_mappers()
# until construction succeeds)
_COMPILE_MUTEX.acquire()
try:
event_registry.add_listener('on_load', reconstruct)
manager.info[_INSTRUMENTOR] = self
-
+
+ @util.deprecated("0.7", message=":meth:`.Mapper.compile` "
+ "is replaced by :func:`.configure_mappers`")
+ def compile(self):
+ """Initialize the inter-mapper relationships of all mappers that
+ have been constructed thus far.
+
+ """
+ configure_mappers()
+ return self
+
+
+ @property
+ @util.deprecated("0.7", message=":attr:`.Mapper.compiled` "
+ "is replaced by :attr:`.Mapper.configured`")
+ def compiled(self):
+ return self.configured
+
def dispose(self):
# Disable any attribute-based compilation.
- self.compiled = True
+ self.configured = True
- if hasattr(self, '_compile_failed'):
- del self._compile_failed
+ if hasattr(self, '_configure_failed'):
+ del self._configure_failed
if not self.non_primary and \
self.class_manager.is_mapped and \
prop.post_instrument_class(self)
- def compile(self):
- """Compile this mapper and all other non-compiled mappers.
-
- This method checks the local compiled status as well as for
- any new mappers that have been defined, and is safe to call
- repeatedly.
-
- """
- global _new_mappers
- if self.compiled and not _new_mappers:
- return self
-
- _COMPILE_MUTEX.acquire()
- try:
- try:
- global _already_compiling
- if _already_compiling:
- return
- _already_compiling = True
- try:
-
- # double-check inside mutex
- if self.compiled and not _new_mappers:
- return self
-
- # initialize properties on all mappers
- # note that _mapper_registry is unordered, which
- # may randomly conceal/reveal issues related to
- # the order of mapper compilation
- for mapper in list(_mapper_registry):
- if getattr(mapper, '_compile_failed', False):
- e = sa_exc.InvalidRequestError(
- "One or more mappers failed to initialize - "
- "can't proceed with initialization of other "
- "mappers. Original exception was: %s"
- % mapper._compile_failed)
- e._compile_failed = mapper._compile_failed
- raise e
- if not mapper.compiled:
- mapper._post_configure_properties()
-
- _new_mappers = False
- return self
- finally:
- _already_compiling = False
- except:
- exc = sys.exc_info()[1]
- if not hasattr(exc, '_compile_failed'):
- self._compile_failed = exc
- raise
- finally:
- self._expire_memoizations()
- _COMPILE_MUTEX.release()
-
def _post_configure_properties(self):
"""Call the ``init()`` method on all ``MapperProperties``
attached to this mapper.
prop.post_instrument_class(self)
self._log("_post_configure_properties() complete")
- self.compiled = True
+ self.configured = True
def add_properties(self, dict_of_properties):
"""Add the given dictionary of properties to this mapper,
def add_property(self, key, prop):
"""Add an individual MapperProperty to this mapper.
- If the mapper has not been compiled yet, just adds the
+ If the mapper has not been configured yet, just adds the
property to the initial properties dictionary sent to the
- constructor. If this Mapper has already been compiled, then
- the given MapperProperty is compiled immediately.
+ constructor. If this Mapper has already been configured, then
+ the given MapperProperty is configured immediately.
"""
self._init_properties[key] = prop
- self._configure_property(key, prop, init=self.compiled)
+ self._configure_property(key, prop, init=self.configured)
self._expire_memoizations()
def _expire_memoizations(self):
for mapper in self.iterate_to_root():
- _memoized_compiled_property.expire_instance(mapper)
+ _memoized_configured_property.expire_instance(mapper)
def _log(self, msg, *args):
self.logger.info(
resolve_synonyms=False and raiseerr=False are deprecated.
"""
-
- if _compile_mappers and not self.compiled:
- self.compile()
+
+ if _compile_mappers and _new_mappers:
+ configure_mappers()
if not resolve_synonyms:
prop = self._props.get(key, None)
@property
def iterate_properties(self):
"""return an iterator of all MapperProperty objects."""
- if not self.compiled:
- self.compile()
+ if _new_mappers:
+ configure_mappers()
return self._props.itervalues()
def _mappers_from_spec(self, spec, selectable):
return from_obj
- @_memoized_compiled_property
+ @_memoized_configured_property
def _single_table_criterion(self):
if self.single and \
self.inherits and \
else:
return None
- @_memoized_compiled_property
+ @_memoized_configured_property
def _with_polymorphic_mappers(self):
if not self.with_polymorphic:
return [self]
return self._mappers_from_spec(*self.with_polymorphic)
- @_memoized_compiled_property
+ @_memoized_configured_property
def _with_polymorphic_selectable(self):
if not self.with_polymorphic:
return self.mapped_table
else:
return mappers, self._selectable_from_mappers(mappers)
- @_memoized_compiled_property
+ @_memoized_configured_property
def _polymorphic_properties(self):
return tuple(self._iterate_polymorphic_properties(
self._with_polymorphic_mappers))
"provided by the get_property() and iterate_properties "
"accessors.")
- @_memoized_compiled_property
+ @_memoized_configured_property
def _get_clause(self):
"""create a "get clause" based on the primary key. this is used
by query.get() and many-to-one lazyloads to load this item
return sql.and_(*[k==v for (k, v) in params]), \
util.column_dict(params)
- @_memoized_compiled_property
+ @_memoized_configured_property
def _equivalent_columns(self):
"""Create a map of all *equivalent* columns, based on
the determination of column pairs that are equated to
yield m
m = m.inherits
- @_memoized_compiled_property
+ @_memoized_configured_property
def self_and_descendants(self):
"""The collection including this mapper and all descendant mappers.
visitables.append((deque(instance_mapper._props.values()),
prp, corresponding_state))
- @_memoized_compiled_property
+ @_memoized_configured_property
def _compiled_cache(self):
return util.LRUCache(self._compiled_cache_size)
- @_memoized_compiled_property
+ @_memoized_configured_property
def _sorted_tables(self):
table_to_mapper = {}
for mapper in self.base_mapper.self_and_descendants:
log.class_logger(Mapper)
+def configure_mappers():
+ """Initialize the inter-mapper relationships of all mappers that
+ have been constructed thus far.
+
+ This function can be called any number of times, but in
+ most cases is handled internally.
+
+ """
+
+ global _new_mappers
+ if not _new_mappers:
+ return
+
+ _COMPILE_MUTEX.acquire()
+ try:
+ global _already_compiling
+ if _already_compiling:
+ return
+ _already_compiling = True
+ try:
+
+ # double-check inside mutex
+ if not _new_mappers:
+ return
+
+ # initialize properties on all mappers
+ # note that _mapper_registry is unordered, which
+ # may randomly conceal/reveal issues related to
+ # the order of mapper compilation
+ for mapper in list(_mapper_registry):
+ if getattr(mapper, '_configure_failed', False):
+ e = sa_exc.InvalidRequestError(
+ "One or more mappers failed to initialize - "
+ "can't proceed with initialization of other "
+ "mappers. Original exception was: %s"
+ % mapper._configure_failed)
+ e._configure_failed = mapper._configure_failed
+ raise e
+ if not mapper.configured:
+ try:
+ mapper._post_configure_properties()
+ mapper._expire_memoizations()
+ except:
+ exc = sys.exc_info()[1]
+ if not hasattr(exc, '_configure_failed'):
+ mapper._configure_failed = exc
+ raise
+
+ _new_mappers = False
+ finally:
+ _already_compiling = False
+ finally:
+ _COMPILE_MUTEX.release()
+
def reconstructor(fn):
"""Decorate a method as the 'reconstructor' hook.
"""Trigger mapper compilation and run init_instance hooks."""
instrumenting_mapper = state.manager.info[_INSTRUMENTOR]
- # compile() always compiles all mappers
- instrumenting_mapper.compile()
+ if _new_mappers:
+ configure_mappers()
if 'init_instance' in instrumenting_mapper.extension:
instrumenting_mapper.extension.init_instance(
instrumenting_mapper, instrumenting_mapper.class_,
join_condition
from sqlalchemy.sql import operators, expression
from sqlalchemy.orm import attributes, dependency, mapper, \
- object_mapper, strategies
+ object_mapper, strategies, configure_mappers
from sqlalchemy.orm.util import CascadeOptions, _class_to_mapper, \
_orm_annotate, _orm_deannotate
from sqlalchemy.orm.interfaces import MANYTOMANY, MANYTOONE, \
MapperProperty, ONETOMANY, PropComparator, StrategizedProperty
+mapperlib = util.importlater("sqlalchemy.orm", "mapperlib")
NoneType = type(None)
__all__ = ('ColumnProperty', 'CompositeProperty', 'SynonymProperty',
@util.memoized_property
def property(self):
- self.prop.parent.compile()
+ if mapperlib.module._new_mappers:
+ configure_mappers()
return self.prop
def compare(self, op, value,
from sqlalchemy.orm.attributes import PASSIVE_NO_RESULT, PASSIVE_OFF, \
NEVER_SET, NO_VALUE, manager_of_class, \
ATTR_WAS_SET
-from sqlalchemy.orm import attributes, exc as orm_exc, interfaces
+from sqlalchemy.orm import attributes, exc as orm_exc, interfaces, configure_mappers
import sys
attributes.state = sys.modules['sqlalchemy.orm.state']
"Cannot deserialize object of type %r - no mapper() has"
" been configured for this class within the current Python process!" %
self.class_)
- elif manager.is_mapped and not manager.mapper.compiled:
- manager.mapper.compile()
+ elif manager.is_mapped and not manager.mapper.configured:
+ configure_mappers()
self.committed_state = state.get('committed_state', {})
self.pending = state.get('pending', {})
else:
return None, entity, False
- if compile:
- mapper = mapper.compile()
+ if compile and mapperlib.module._new_mappers:
+ mapperlib.configure_mappers()
return mapper, mapper._with_polymorphic_selectable, False
def _entity_descriptor(entity, key):
except exc.NO_STATE:
raise exc.UnmappedClassError(class_)
- if compile:
- mapper = mapper.compile()
+ if compile and mapperlib.module._new_mappers:
+ mapperlib.configure_mappers()
return mapper
def _class_to_mapper(class_or_mapper, compile=True):
else:
raise exc.UnmappedClassError(class_or_mapper)
- if compile:
- return mapper.compile()
- else:
- return mapper
+ if compile and mapperlib.module._new_mappers:
+ mapperlib.configure_mappers()
+ return mapper
def has_identity(object):
state = attributes.instance_state(object)
self._il_addtl = addtl
@memoized_property
- def _il_module(self):
+ def module(self):
m = __import__(self._il_path)
for token in self._il_path.split(".")[1:]:
m = getattr(m, token)
def __getattr__(self, key):
try:
- attr = getattr(self._il_module, key)
+ attr = getattr(self.module, key)
except AttributeError:
raise AttributeError(
"Module %s has no attribute '%s'" %
ForeignKeyConstraint, asc, Index
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.orm import relationship, create_session, class_mapper, \
- joinedload, compile_mappers, backref, clear_mappers, \
+ joinedload, configure_mappers, backref, clear_mappers, \
polymorphic_union, deferred, column_property
from sqlalchemy.test.testing import eq_
from sqlalchemy.util import classproperty
assert_raises_message(exc.InvalidRequestError,
"'addresses' is not an instance of "
- "ColumnProperty", compile_mappers)
+ "ColumnProperty", configure_mappers)
def test_string_dependency_resolution_two(self):
assert_raises_message(exc.InvalidRequestError,
"does not have a mapped column named "
- "'__table__'", compile_mappers)
+ "'__table__'", configure_mappers)
def test_string_dependency_resolution_no_magic(self):
"""test that full tinkery expressions work as written"""
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
- compile_mappers()
+ configure_mappers()
eq_(str(User.addresses.prop.primaryjoin),
'users.id = addresses.user_id')
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
- compile_mappers()
+ configure_mappers()
eq_(str(User.addresses.property.primaryjoin),
str(Address.user.property.primaryjoin))
Column('user_id', Integer,
ForeignKey('users.id')), Column('prop_id',
Integer, ForeignKey('props.id')))
- compile_mappers()
+ configure_mappers()
assert class_mapper(User).get_property('props').secondary \
is user_to_prop
# this used to raise an error when accessing User.id but that's
# no longer the case since we got rid of _CompileOnAttr.
- assert_raises(sa.exc.ArgumentError, compile_mappers)
+ assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_nice_dependency_error_works_with_hasattr(self):
"^One or more mappers failed to initialize - "
"can't proceed with initialization of other "
"mappers. Original exception was: When initializing.*",
- compile_mappers)
+ configure_mappers)
def test_custom_base(self):
class MyBase(object):
master = relationship(Master)
Base.metadata.create_all()
- compile_mappers()
+ configure_mappers()
assert class_mapper(Detail).get_property('master'
).strategy.use_get
m1 = Master()
i = Index('my_index', User.name)
# compile fails due to the nonexistent Addresses relationship
- assert_raises(sa.exc.InvalidRequestError, compile_mappers)
+ assert_raises(sa.exc.InvalidRequestError, configure_mappers)
# index configured
assert i in User.__table__.indexes
# the User.id inside the ForeignKey but this is no longer the
# case
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
eq_(str(Address.user_id.property.columns[0].foreign_keys[0]),
"ForeignKey('users.id')")
Base.metadata.create_all()
# compile succeeds because inherit_condition is honored
- compile_mappers()
+ configure_mappers()
def test_joined(self):
related_child1 = Column('c1', Integer)
__mapper_args__ = dict(polymorphic_identity='child2')
- sa.orm.compile_mappers() # no exceptions here
+ sa.orm.configure_mappers() # no exceptions here
def test_single_colsonbase(self):
"""test single inheritance where all the columns are on the base
== user_id, backref='addresses')
if not inline:
- compile_mappers()
+ configure_mappers()
if stringbased:
Address.user = relationship('User',
primaryjoin='User.id==Address.user_id',
class Engineer(Person):
pass
- compile_mappers()
+ configure_mappers()
assert class_mapper(Person).polymorphic_on \
is Person.__table__.c.type
eq_(class_mapper(Engineer).polymorphic_identity, 'Engineer')
class Engineer(Person, ComputedMapperArgs):
pass
- compile_mappers()
+ configure_mappers()
assert class_mapper(Person).polymorphic_on \
is Person.__table__.c.type
eq_(class_mapper(Engineer).polymorphic_identity, 'Engineer')
__tablename__ = 'test'
id = Column(Integer, primary_key=True)
- compile_mappers()
+ configure_mappers()
eq_(MyModel.type_.__doc__, """this is a document.""")
eq_(MyModel.t2.__doc__, """this is another document.""")
__tablename__ = 'test'
id = Column(Integer, primary_key=True)
- compile_mappers()
+ configure_mappers()
col = MyModel.__mapper__.polymorphic_on
eq_(col.name, 'type_')
assert col.table is not None
from sqlalchemy.test.schema import Table
from sqlalchemy.test.schema import Column
from sqlalchemy.orm import relationship, sessionmaker, scoped_session, \
- class_mapper, mapper, joinedload, compile_mappers, aliased
+ class_mapper, mapper, joinedload, configure_mappers, aliased
from sqlalchemy.test.testing import eq_
from test.orm._base import ComparableEntity, MappedTest
: relationship(Address, backref='user',
order_by=addresses.c.id)})
mapper(Address, addresses)
- compile_mappers()
+ configure_mappers()
@classmethod
def insert_data(cls):
polymorphic_on=t1t2_join.c.x,
with_polymorphic=('*', t1t2_join),
polymorphic_identity=0)
- compile_mappers()
+ configure_mappers()
clear_mappers()
'data':relationship(mapper(Data, data))
},
order_by=table1.c.id)
- table1_mapper.compile()
+ configure_mappers()
assert False
except:
assert True
table3_mapper = mapper(Table3, table3, inherits=table1_mapper, polymorphic_identity='table3')
- table1_mapper.compile()
+ configure_mappers()
assert table1_mapper.primary_key == [table1.c.id], table1_mapper.primary_key
@testing.fails_on('maxdb', 'FIXME: unknown')
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import *
from sqlalchemy.test import *
+from sqlalchemy.test.testing import assert_raises_message
from test.orm import _base
def teardown(self):
clear_mappers()
- def testone(self):
+ def test_with_polymorphic(self):
metadata = MetaData(testing.db)
order = Table('orders', metadata,
# this requires that the compilation of order_mapper's "surrogate
# mapper" occur after the initial setup of MapperProperty objects on
# the mapper.
- class_mapper(Product).compile()
+ configure_mappers()
- def testtwo(self):
+ def test_conflicting_backref_one(self):
"""test that conflicting backrefs raises an exception"""
metadata = MetaData(testing.db)
mapper(OrderProduct, orderproduct)
- try:
- class_mapper(Product).compile()
- assert False
- except sa_exc.ArgumentError, e:
- assert str(e).index("Error creating backref ") > -1
+ assert_raises_message(
+ sa_exc.ArgumentError,
+ "Error creating backref",
+ configure_mappers
+ )
- def testthree(self):
+ def test_misc_one(self):
metadata = MetaData(testing.db)
node_table = Table("node", metadata,
Column('node_id', Integer, primary_key=True),
finally:
metadata.drop_all()
- def testfour(self):
+ def test_conflicting_backref_two(self):
meta = MetaData()
a = Table('a', meta, Column('id', Integer, primary_key=True))
- b = Table('b', meta, Column('id', Integer, primary_key=True), Column('a_id', Integer, ForeignKey('a.id')))
+ b = Table('b', meta, Column('id', Integer, primary_key=True),
+ Column('a_id', Integer, ForeignKey('a.id')))
class A(object):pass
class B(object):pass
'a':relationship(A, backref='b')
})
- try:
- compile_mappers()
- assert False
- except sa_exc.ArgumentError, e:
- assert str(e).index("Error creating backref") > -1
-
+ assert_raises_message(
+ sa_exc.ArgumentError,
+ "Error creating backref",
+ configure_mappers
+ )
'places':relationship(Place, secondary=place_input, backref='transitions')
})
assert_raises_message(sa.exc.ArgumentError, "Error creating backref",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_circular(self):
from sqlalchemy import MetaData, Integer, String, ForeignKey, func, util
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.engine import default
-from sqlalchemy.orm import mapper, relationship, backref, create_session, class_mapper, compile_mappers, reconstructor, validates, aliased
+from sqlalchemy.orm import mapper, relationship, backref, create_session, class_mapper, configure_mappers, reconstructor, validates, aliased
from sqlalchemy.orm import defer, deferred, synonym, attributes, column_property, composite, relationship, dynamic_loader, comparable_property,AttributeExtension
from sqlalchemy.test.testing import eq_, AssertsCompiledSQL
from test.orm import _base, _fixtures
properties={
'addresses':relationship(Address, backref='email_address')
})
- assert_raises(sa.exc.ArgumentError, sa.orm.compile_mappers)
+ assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_update_attr_keys(self):
"initialization of other mappers. "
"Original exception was: Class "
"'test.orm._fixtures.User' is not mapped$"
- , compile_mappers)
+ , configure_mappers)
@testing.resolve_artifact_names
def test_column_prefix(self):
assert_raises(sa.exc.ArgumentError, mapper, User, s)
@testing.resolve_artifact_names
- def test_recompile_on_other_mapper(self):
- """A compile trigger on an already-compiled mapper still triggers a check against all mappers."""
+ def test_reconfigure_on_other_mapper(self):
+ """A configure trigger on an already-configured mapper
+ still triggers a check against all mappers."""
mapper(User, users)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
assert sa.orm.mapperlib._new_mappers is False
m = mapper(Address, addresses, properties={
'user': relationship(User, backref="addresses")})
- assert m.compiled is False
+ assert m.configured is False
assert sa.orm.mapperlib._new_mappers is True
u = User()
assert User.addresses
assert sa.orm.mapperlib._new_mappers is False
@testing.resolve_artifact_names
- def test_compile_on_session(self):
+ def test_configure_on_session(self):
m = mapper(User, users)
session = create_session()
session.connection(m)
def test_props(self):
m = mapper(User, users, properties = {
'addresses' : relationship(mapper(Address, addresses))
- }).compile()
+ })
assert User.addresses.property is m.get_property('addresses')
@testing.resolve_artifact_names
- def test_compile_on_prop_1(self):
+ def test_configure_on_prop_1(self):
mapper(User, users, properties = {
'addresses' : relationship(mapper(Address, addresses))
})
User.addresses.any(Address.email_address=='foo@bar.com')
@testing.resolve_artifact_names
- def test_compile_on_prop_2(self):
+ def test_configure_on_prop_2(self):
mapper(User, users, properties = {
'addresses' : relationship(mapper(Address, addresses))
})
eq_(str(User.id == 3), str(users.c.id==3))
@testing.resolve_artifact_names
- def test_compile_on_prop_3(self):
+ def test_configure_on_prop_3(self):
class Foo(User):pass
mapper(User, users)
mapper(Foo, addresses, inherits=User)
def test_deferred_subclass_attribute_instrument(self):
class Foo(User):pass
mapper(User, users)
- compile_mappers()
+ configure_mappers()
mapper(Foo, addresses, inherits=User)
assert getattr(Foo().__class__, 'name').impl is not None
class Foo(User):pass
m = mapper(User, users)
mapper(Order, orders)
- compile_mappers()
+ configure_mappers()
mapper(Foo, addresses, inherits=User)
ext_list = [AttributeExtension()]
m.add_property('somename', column_property(users.c.name, extension=ext_list))
assert Foo.orders.impl.extensions is User.orders.impl.extensions
assert Foo.orders.impl.extensions is not ext_list
- compile_mappers()
+ configure_mappers()
assert len(User.somename.impl.extensions) == 1
assert len(Foo.somename.impl.extensions) == 1
assert len(Foo.orders.impl.extensions) == 3
@testing.resolve_artifact_names
- def test_compile_on_get_props_1(self):
+ def test_configure_on_get_props_1(self):
m =mapper(User, users)
- assert not m.compiled
+ assert not m.configured
assert list(m.iterate_properties)
- assert m.compiled
+ assert m.configured
@testing.resolve_artifact_names
- def test_compile_on_get_props_2(self):
+ def test_configure_on_get_props_2(self):
m= mapper(User, users)
- assert not m.compiled
+ assert not m.configured
assert m.get_property('name')
- assert m.compiled
+ assert m.configured
+
+ @testing.resolve_artifact_names
+ def test_configure_on_get_props_3(self):
+ m= mapper(User, users)
+ assert not m.configured
+ configure_mappers()
+
+ m2 = mapper(Address, addresses, properties={
+ 'user':relationship(User, backref='addresses')
+ })
+ assert m.get_property('addresses')
@testing.resolve_artifact_names
def test_add_property(self):
mapper(Address, addresses, properties={
'user':synonym('_user')
})
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
# later, backref sets up the prop
mapper(User, users, properties={
def test_illegal_non_primary(self):
mapper(User, users)
mapper(Address, addresses)
+ mapper(User, users, non_primary=True, properties={
+ 'addresses':relationship(Address)
+ })
assert_raises_message(
sa.exc.ArgumentError,
"Attempting to assign a new relationship 'addresses' "
"to a non-primary mapper on class 'User'",
- mapper(User, users, non_primary=True, properties={
- 'addresses':relationship(Address)
- }).compile
+ configure_mappers
)
@testing.resolve_artifact_names
exclude_properties=(t.c.boss_id,
'employee_number', t.c.vendor_id))
- p_m.compile()
+ configure_mappers()
def assert_props(cls, want):
have = set([n for n in dir(cls) if not n.startswith('_')])
addresses.join(email_bounces),
properties={'id':[addresses.c.id, email_bounces.c.id]}
)
- m.compile()
+ configure_mappers()
assert addresses in m._pks_by_table
assert email_bounces not in m._pks_by_table
class MyFakeProperty(sa.orm.properties.ColumnProperty):
def post_instrument_class(self, mapper):
super(MyFakeProperty, self).post_instrument_class(mapper)
- m2.compile()
+ configure_mappers()
m1 = mapper(User, users, properties={
'name':MyFakeProperty(users.c.name)
})
m2 = mapper(Address, addresses)
- compile_mappers()
+ configure_mappers()
sa.orm.clear_mappers()
class MyFakeProperty(sa.orm.properties.ColumnProperty):
def post_instrument_class(self, mapper):
super(MyFakeProperty, self).post_instrument_class(mapper)
- m1.compile()
+ configure_mappers()
m1 = mapper(User, users, properties={
'name':MyFakeProperty(users.c.name)
})
m2 = mapper(Address, addresses)
- compile_mappers()
+ configure_mappers()
@testing.resolve_artifact_names
def test_reconstructor(self):
'addresses':relationship(Address)
})
- assert_raises(sa.orm.exc.UnmappedClassError, sa.orm.compile_mappers)
+ assert_raises(sa.orm.exc.UnmappedClassError, sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_unmapped_subclass_error_postmap(self):
pass
mapper(Base, users)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
# we can create new instances, set attributes.
s = Sub()
class Sub(Base):
pass
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
# we can create new instances, set attributes.
s = Sub()
'hoho':synonym(t1.c.col4, doc="syn of col4")
})
mapper(Bar, t2)
- compile_mappers()
+ configure_mappers()
eq_(Foo.col1.__doc__, "primary key column")
eq_(Foo.col2.__doc__, "data col")
eq_(Foo.col5.__doc__, None)
from sqlalchemy.orm import mapper, relationship, create_session, \
sessionmaker, attributes, interfaces,\
clear_mappers, exc as orm_exc,\
- compile_mappers
+ configure_mappers
from test.orm import _base, _fixtures
mapper(CompositePk, composite_pk_table)
- compile_mappers()
+ configure_mappers()
class RowTupleTest(QueryTest):
run_setup_mappers = None
from sqlalchemy import Integer, String, ForeignKey, MetaData, and_
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, relation, \
- backref, create_session, compile_mappers, clear_mappers, sessionmaker
+ backref, create_session, configure_mappers, clear_mappers, sessionmaker
from sqlalchemy.test.testing import eq_, startswith_
from test.orm import _base, _fixtures
'b':relationship(B, cascade="all,delete-orphan", uselist=False)})
mapper(B, tableB)
- compile_mappers()
+ configure_mappers()
assert A.b.property.strategy.use_get
sess = create_session()
'user':relationship(User, back_populates='addresses')
})
- assert_raises(sa.exc.InvalidRequestError, compile_mappers)
+ assert_raises(sa.exc.InvalidRequestError, configure_mappers)
@testing.resolve_artifact_names
def test_invalid_target(self):
assert_raises_message(sa.exc.ArgumentError,
r"reverse_property 'dingaling' on relationship User.addresses references "
"relationship Address.dingaling, which does not reference mapper Mapper\|User\|users",
- compile_mappers)
+ configure_mappers)
class JoinConditionErrorTest(testing.TestBase):
c1id = Column('c1id', Integer, ForeignKey('c1.id'))
c2 = relationship(C1, primaryjoin=C1.id)
- assert_raises(sa.exc.ArgumentError, compile_mappers)
+ assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_clauseelement_pj_false(self):
from sqlalchemy.ext.declarative import declarative_base
c1id = Column('c1id', Integer, ForeignKey('c1.id'))
c2 = relationship(C1, primaryjoin="x"=="y")
- assert_raises(sa.exc.ArgumentError, compile_mappers)
+ assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_only_column_elements(self):
m = MetaData()
mapper(C1, t1, properties={'c2':relationship(C2, primaryjoin=t1.join(t2))})
mapper(C2, t2)
- assert_raises(sa.exc.ArgumentError, compile_mappers)
+ assert_raises(sa.exc.ArgumentError, configure_mappers)
def test_invalid_string_args(self):
from sqlalchemy.ext.declarative import declarative_base
assert_raises_message(
sa.exc.ArgumentError,
"Column-based expression object expected for argument '%s'; got: '%s', type %r" % (argname, arg[0], type(arg[0])),
- compile_mappers)
+ configure_mappers)
def test_fk_error_raised(self):
mapper(C1, t1, properties={'c2':relationship(C2)})
mapper(C2, t3)
- assert_raises(sa.exc.NoReferencedColumnError, compile_mappers)
+ assert_raises(sa.exc.NoReferencedColumnError, configure_mappers)
def test_join_error_raised(self):
m = MetaData()
mapper(C1, t1, properties={'c2':relationship(C2)})
mapper(C2, t3)
- assert_raises(sa.exc.ArgumentError, compile_mappers)
+ assert_raises(sa.exc.ArgumentError, configure_mappers)
def teardown(self):
clear_mappers()
b_plain= relationship( B, secondary=t12),
)
)
- compile_mappers()
+ configure_mappers()
assert m.get_property('b_view').local_remote_pairs == \
m.get_property('b_plain').local_remote_pairs == \
[(t1.c.id, t12.c.t1_id), (t2.c.id, t12.c.t2_id)]
mapper(T3, t3)
assert_raises_message(sa.exc.ArgumentError,
"Specify remote_side argument",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
class ExplicitLocalRemoteTest(_base.MappedTest):
foreign_keys=[t2.c.t1id],
remote_side=[t2.c.t1id])})
mapper(T2, t2)
- assert_raises(sa.exc.ArgumentError, sa.orm.compile_mappers)
+ assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_escalation_2(self):
primaryjoin=t1.c.id==sa.func.lower(t2.c.t1id),
_local_remote_pairs=[(t1.c.id, t2.c.t1id)])})
mapper(T2, t2)
- assert_raises(sa.exc.ArgumentError, sa.orm.compile_mappers)
+ assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
class InvalidRemoteSideTest(_base.MappedTest):
@classmethod
assert_raises_message(sa.exc.ArgumentError, "T1.t1s and back-reference T1.parent are "
"both of the same direction <symbol 'ONETOMANY>. Did you "
- "mean to set remote_side on the many-to-one side ?", sa.orm.compile_mappers)
+ "mean to set remote_side on the many-to-one side ?", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_m2o_backref(self):
assert_raises_message(sa.exc.ArgumentError, "T1.t1s and back-reference T1.parent are "
"both of the same direction <symbol 'MANYTOONE>. Did you "
- "mean to set remote_side on the many-to-one side ?", sa.orm.compile_mappers)
+ "mean to set remote_side on the many-to-one side ?", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_o2m_explicit(self):
# can't be sure of ordering here
assert_raises_message(sa.exc.ArgumentError,
"both of the same direction <symbol 'ONETOMANY>. Did you "
- "mean to set remote_side on the many-to-one side ?", sa.orm.compile_mappers)
+ "mean to set remote_side on the many-to-one side ?", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_m2o_explicit(self):
# can't be sure of ordering here
assert_raises_message(sa.exc.ArgumentError,
"both of the same direction <symbol 'MANYTOONE>. Did you "
- "mean to set remote_side on the many-to-one side ?", sa.orm.compile_mappers)
+ "mean to set remote_side on the many-to-one side ?", sa.orm.configure_mappers)
class InvalidRelationshipEscalationTest(_base.MappedTest):
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine join condition between parent/child "
- "tables on relationship", sa.orm.compile_mappers)
+ "tables on relationship", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_join_self_ref(self):
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine join condition between parent/child "
- "tables on relationship", sa.orm.compile_mappers)
+ "tables on relationship", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_equated(self):
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine relationship direction for primaryjoin condition",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_equated_fks(self):
assert_raises_message(
sa.exc.ArgumentError,
"Could not locate any equated, locally mapped column pairs "
- "for primaryjoin condition", sa.orm.compile_mappers)
+ "for primaryjoin condition", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_ambiguous_fks(self):
"ForeignKeyConstraint objects "
r"established \(in which case "
r"'foreign_keys' is usually unnecessary\)\?"
- , sa.orm.compile_mappers)
+ , sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_ambiguous_remoteside_o2m(self):
assert_raises_message(
sa.exc.ArgumentError,
"could not determine any local/remote column pairs",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_ambiguous_remoteside_m2o(self):
assert_raises_message(
sa.exc.ArgumentError,
"could not determine any local/remote column pairs",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine relationship direction for primaryjoin condition",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_equated_self_ref(self):
assert_raises_message(
sa.exc.ArgumentError,
"Could not locate any equated, locally mapped column pairs "
- "for primaryjoin condition", sa.orm.compile_mappers)
+ "for primaryjoin condition", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_equated_viewonly(self):
assert_raises_message(sa.exc.ArgumentError,
'Could not determine relationship '
'direction for primaryjoin condition',
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
sa.orm.clear_mappers()
mapper(Foo, foos_with_fks, properties={
primaryjoin=foos_with_fks.c.id>bars_with_fks.c.fid,
viewonly=True)})
mapper(Bar, bars_with_fks)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
@testing.resolve_artifact_names
def test_no_equated_self_ref_viewonly(self):
"Column objects have a ForeignKey "
"present, or are otherwise part of a "
"ForeignKeyConstraint on their parent "
- "Table.", sa.orm.compile_mappers)
+ "Table.", sa.orm.configure_mappers)
sa.orm.clear_mappers()
mapper(Foo, foos_with_fks, properties={
primaryjoin=foos_with_fks.c.id>foos_with_fks.c.fid,
viewonly=True)})
mapper(Bar, bars_with_fks)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
@testing.resolve_artifact_names
def test_no_equated_self_ref_viewonly_fks(self):
viewonly=True,
foreign_keys=[foos.c.fid])})
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
eq_(Foo.foos.property.local_remote_pairs, [(foos.c.id, foos.c.fid)])
@testing.resolve_artifact_names
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine relationship direction for primaryjoin condition",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
sa.orm.clear_mappers()
mapper(Foo, foos_with_fks, properties={
'bars':relationship(Bar,
primaryjoin=foos_with_fks.c.id==bars_with_fks.c.fid)})
mapper(Bar, bars_with_fks)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
@testing.resolve_artifact_names
def test_equated_self_ref(self):
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine relationship direction for primaryjoin condition",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine relationship direction for primaryjoin condition",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
class InvalidRelationshipEscalationTestM2M(_base.MappedTest):
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine join condition between parent/child tables "
- "on relationship", sa.orm.compile_mappers)
+ "on relationship", sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_secondaryjoin(self):
sa.exc.ArgumentError,
"Could not determine join condition between parent/child tables "
"on relationship",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_fks_warning_1(self):
"'foobars.bid', 'foobars.fid' for join "
"condition 'foos.id = foobars.fid' on "
"relationship Foo.bars",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
sa.orm.clear_mappers()
mapper(Foo, foos, properties={
"join condition 'foos.id = "
"foobars_with_many_columns.fid' on "
"relationship Foo.bars",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
@testing.emits_warning(r'No ForeignKey objects.*')
@testing.resolve_artifact_names
primaryjoin=foos.c.id==foobars.c.fid,
secondaryjoin=foobars.c.bid==bars.c.id)})
mapper(Bar, bars)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
eq_(
Foo.bars.property.synchronize_pairs,
[(foos.c.id, foobars.c.fid)]
primaryjoin=foos.c.id==foobars_with_many_columns.c.fid,
secondaryjoin=foobars_with_many_columns.c.bid==bars.c.id)})
mapper(Bar, bars)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
eq_(
Foo.bars.property.synchronize_pairs,
[(foos.c.id, foobars_with_many_columns.c.fid)]
assert_raises_message(
sa.exc.ArgumentError,
"Could not determine relationship direction for primaryjoin condition",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
sa.orm.clear_mappers()
mapper(Foo, foos, properties={
assert_raises_message(
sa.exc.ArgumentError,
"Could not locate any equated, locally mapped column pairs for primaryjoin condition ",
- sa.orm.compile_mappers)
+ sa.orm.configure_mappers)
sa.orm.clear_mappers()
mapper(Foo, foos, properties={
secondaryjoin=foobars_with_fks.c.bid<=bars.c.id,
viewonly=True)})
mapper(Bar, bars)
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
@testing.resolve_artifact_names
def test_bad_secondaryjoin(self):
"ForeignKey and/or ForeignKeyConstraint "
r"objects established \(in which case "
r"'foreign_keys' is usually unnecessary\)?"
- , sa.orm.compile_mappers)
+ , sa.orm.configure_mappers)
@testing.resolve_artifact_names
def test_no_equated_secondaryjoin(self):
assert_raises_message(
sa.exc.ArgumentError,
"Could not locate any equated, locally mapped column pairs for "
- "secondaryjoin condition", sa.orm.compile_mappers)
+ "secondaryjoin condition", sa.orm.configure_mappers)
class RelationDeprecationTest(_base.MappedTest):
m2 = mapper(Person, people, properties={
'sites' : relationship(PersonSite)})
- sa.orm.compile_mappers()
+ sa.orm.configure_mappers()
eq_(list(m2.get_property('sites').synchronize_pairs),
[(people.c.person, peoplesites.c.person)])
'myclass':relationship(MyClass, cascade="all, delete", passive_deletes=True)
})
mapper(MyClass, mytable)
- assert_raises(sa.exc.SAWarning, sa.orm.compile_mappers)
+ assert_raises(sa.exc.SAWarning, sa.orm.configure_mappers)
class ExtraPassiveDeletesTest(_base.MappedTest):
__requires__ = ('foreign_keys',)