from ...schema import Table, MetaData
from ...orm import synonym as _orm_synonym, mapper,\
comparable_property,\
- interfaces
+ interfaces, properties
from ...orm.util import polymorphic_union
from ...orm.base import _mapper_or_none
+from ...util import compat
from ... import exc
import weakref
from .base import _as_declarative, \
_declarative_constructor,\
_MapperConfig, _add_attribute
+from .clsregistry import _class_resolver
def instrument_declarative(cls, registry, metadata):
def prepare(cls, engine):
"""Reflect all :class:`.Table` objects for all current
:class:`.DeferredReflection` subclasses"""
+
to_map = [m for m in _MapperConfig.configs.values()
if issubclass(m.cls, cls)]
for thingy in to_map:
cls._sa_decl_prepare(thingy.local_table, engine)
thingy.map()
+ mapper = thingy.cls.__mapper__
+ metadata = mapper.class_.metadata
+ for rel in mapper._props.values():
+ if isinstance(rel, properties.RelationshipProperty) and \
+ rel.secondary is not None:
+ if isinstance(rel.secondary, Table):
+ cls._sa_decl_prepare(rel.secondary, engine)
+ elif isinstance(rel.secondary, _class_resolver):
+ rel.secondary._resolvers += (
+ cls._sa_deferred_table_resolver(engine, metadata),
+ )
+
+ @classmethod
+ def _sa_deferred_table_resolver(cls, engine, metadata):
+ def _resolve(key):
+ t1 = Table(key, metadata)
+ cls._sa_decl_prepare(t1, engine)
+ return t1
+ return _resolve
@classmethod
def _sa_decl_prepare(cls, local_table, engine):
return _GetColumns(value)
-def _resolver(cls, prop):
- def resolve_arg(arg):
- import sqlalchemy
- from sqlalchemy.orm import foreign, remote
-
- fallback = sqlalchemy.__dict__.copy()
- fallback.update({'foreign': foreign, 'remote': remote})
-
- def access_cls(key):
- if key in cls._decl_class_registry:
- return _determine_container(key, cls._decl_class_registry[key])
- elif key in cls.metadata.tables:
- return cls.metadata.tables[key]
- elif key in cls.metadata._schemas:
- return _GetTable(key, cls.metadata)
- elif '_sa_module_registry' in cls._decl_class_registry and \
- key in cls._decl_class_registry['_sa_module_registry']:
- registry = cls._decl_class_registry['_sa_module_registry']
- return registry.resolve_attr(key)
+class _class_resolver(object):
+ def __init__(self, cls, prop, fallback, arg):
+ self.cls = cls
+ self.prop = prop
+ self.arg = self._declarative_arg = arg
+ self.fallback = fallback
+ self._dict = util.PopulateDict(self._access_cls)
+ self._resolvers = ()
+
+ def _access_cls(self, key):
+ cls = self.cls
+ if key in cls._decl_class_registry:
+ return _determine_container(key, cls._decl_class_registry[key])
+ elif key in cls.metadata.tables:
+ return cls.metadata.tables[key]
+ elif key in cls.metadata._schemas:
+ return _GetTable(key, cls.metadata)
+ elif '_sa_module_registry' in cls._decl_class_registry and \
+ key in cls._decl_class_registry['_sa_module_registry']:
+ registry = cls._decl_class_registry['_sa_module_registry']
+ return registry.resolve_attr(key)
+ elif self._resolvers:
+ for resolv in self._resolvers:
+ value = resolv(key)
+ if value is not None:
+ return value
+
+ return self.fallback[key]
+
+ def __call__(self):
+ try:
+ x = eval(self.arg, globals(), self._dict)
+
+ if isinstance(x, _GetColumns):
+ return x.cls
else:
- return fallback[key]
+ return x
+ except NameError as n:
+ raise exc.InvalidRequestError(
+ "When initializing mapper %s, expression %r failed to "
+ "locate a name (%r). If this is a class name, consider "
+ "adding this relationship() to the %r class after "
+ "both dependent classes have been defined." %
+ (self.prop.parent, self.arg, n.args[0], self.cls)
+ )
- d = util.PopulateDict(access_cls)
- def return_cls():
- try:
- x = eval(arg, globals(), d)
+def _resolver(cls, prop):
+ import sqlalchemy
+ from sqlalchemy.orm import foreign, remote
- if isinstance(x, _GetColumns):
- return x.cls
- else:
- return x
- except NameError as n:
- raise exc.InvalidRequestError(
- "When initializing mapper %s, expression %r failed to "
- "locate a name (%r). If this is a class name, consider "
- "adding this relationship() to the %r class after "
- "both dependent classes have been defined." %
- (prop.parent, arg, n.args[0], cls)
- )
- return return_cls
+ fallback = sqlalchemy.__dict__.copy()
+ fallback.update({'foreign': foreign, 'remote': remote})
+
+ def resolve_arg(arg):
+ return _class_resolver(cls, prop, fallback, arg)
return resolve_arg
test_needs_fk=True,
)
- def test_basic(self):
- meta = MetaData(testing.db)
+ def test_basic(self):
class User(Base, fixtures.ComparableEntity):
__tablename__ = 'users'
eq_(a1.user, User(name='u1'))
def test_rekey(self):
- meta = MetaData(testing.db)
-
class User(Base, fixtures.ComparableEntity):
__tablename__ = 'users'
assert_raises(TypeError, User, name='u3')
def test_supplied_fk(self):
- meta = MetaData(testing.db)
-
class IMHandle(Base, fixtures.ComparableEntity):
__tablename__ = 'imhandles'
class DeferredReflectBase(DeclarativeReflectionBase):
def teardown(self):
- super(DeferredReflectBase,self).teardown()
+ super(DeferredReflectBase, self).teardown()
from sqlalchemy.ext.declarative.base import _MapperConfig
_MapperConfig.configs.clear()
@decl.declared_attr
def __mapper_args__(cls):
return {
- "order_by":cls.__table__.c.name
+ "order_by": cls.__table__.c.name
}
decl.DeferredReflection.prepare(testing.db)
]
)
+class DeferredSecondaryReflectionTest(DeferredReflectBase):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('users', metadata,
+ Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(50)), test_needs_fk=True)
+
+ Table('user_items', metadata,
+ Column('user_id', ForeignKey('users.id'), primary_key=True),
+ Column('item_id', ForeignKey('items.id'), primary_key=True),
+ test_needs_fk=True
+ )
+
+ Table('items', metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('name', String(50)),
+ test_needs_fk=True
+ )
+
+ def _roundtrip(self):
+
+ User = Base._decl_class_registry['User']
+ Item = Base._decl_class_registry['Item']
+
+ u1 = User(name='u1', items=[Item(name='i1'), Item(name='i2')])
+
+ sess = Session()
+ sess.add(u1)
+ sess.commit()
+
+ eq_(sess.query(User).all(), [User(name='u1',
+ items=[Item(name='i1'), Item(name='i2')])])
+
+ def test_string_resolution(self):
+ class User(decl.DeferredReflection, fixtures.ComparableEntity, Base):
+ __tablename__ = 'users'
+
+ items = relationship("Item", secondary="user_items")
+
+ class Item(decl.DeferredReflection, fixtures.ComparableEntity, Base):
+ __tablename__ = 'items'
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_table_resolution(self):
+ class User(decl.DeferredReflection, fixtures.ComparableEntity, Base):
+ __tablename__ = 'users'
+
+ items = relationship("Item", secondary=Table("user_items", Base.metadata))
+
+ class Item(decl.DeferredReflection, fixtures.ComparableEntity, Base):
+ __tablename__ = 'items'
+
+ decl.DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
class DeferredInhReflectBase(DeferredReflectBase):
def _roundtrip(self):
Foo = Base._decl_class_registry['Foo']
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
class Bar(Foo):
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
decl.DeferredReflection.prepare(testing.db)
self._roundtrip()
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
class Bar(Foo):
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
bar_data = Column(String(30))
decl.DeferredReflection.prepare(testing.db)
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
id = Column(Integer, primary_key=True)
class Bar(Foo):
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
decl.DeferredReflection.prepare(testing.db)
self._roundtrip()
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__tablename__ = 'bar'
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
decl.DeferredReflection.prepare(testing.db)
self._roundtrip()
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__tablename__ = 'bar'
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
bar_data = Column(String(30))
decl.DeferredReflection.prepare(testing.db)
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
id = Column(Integer, primary_key=True)
class Bar(Foo):
__tablename__ = 'bar'
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
decl.DeferredReflection.prepare(testing.db)
self._roundtrip()
class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
Base):
__tablename__ = 'foo'
- __mapper_args__ = {"polymorphic_on":"type",
- "polymorphic_identity":"foo"}
+ __mapper_args__ = {"polymorphic_on": "type",
+ "polymorphic_identity": "foo"}
class Bar(Foo):
__tablename__ = 'bar'
- __mapper_args__ = {"polymorphic_identity":"bar"}
+ __mapper_args__ = {"polymorphic_identity": "bar"}
id = Column(Integer, ForeignKey('foo.id'), primary_key=True)
decl.DeferredReflection.prepare(testing.db)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
