global object_session
if not object_session:
from sqlalchemy.orm.session import object_session
+
+ session = object_session(instance)
+ mapper = object_mapper(instance)
+ if not session:
+ try:
+ session = mapper.get_session()
+ except exceptions.InvalidRequestError:
+ raise exceptions.InvalidRequestError("Instance %s is not bound to a Session, and no contextual session is established; attribute refresh operation cannot proceed" % (instance.__class__))
- if object_session(instance).query(object_mapper(instance))._get(instance._instance_key, refresh_instance=instance._state, only_load_props=attribute_names) is None:
+ if session.query(mapper)._get(instance._instance_key, refresh_instance=instance._state, only_load_props=attribute_names) is None:
raise exceptions.InvalidRequestError("Could not refresh instance '%s'" % instance_str(instance))
def _state_mapper(state, entity_name=None):
meta.drop_all()
def test_override_create_fkcols(self):
- """test that you can override columns and create new foreign keys to other reflected tables.
- this is common with MySQL MyISAM tables."""
+ """test that you can override columns and create new foreign keys to other reflected tables
+ which have no foreign keys. this is common with MySQL MyISAM tables."""
meta = MetaData(testbase.db)
users = Table('users', meta,
finally:
meta.drop_all()
+
def test_unknown_types(self):
meta = MetaData(testbase.db)
t = Table("test", meta,
assert [c.parent for c in a2.c.user_id.foreign_keys] == [a2.c.user_id]
assert list(a2.c.user_id.foreign_keys)[0].parent is a2.c.user_id
assert u2.join(a2).onclause == u2.c.id==a2.c.user_id
+
+ meta2 = MetaData(testbase.db)
+ u2 = Table('users', meta2,
+ Column('id', Integer, primary_key=True),
+ autoload=True)
+ a2 = Table('addresses', meta2,
+ Column('id', Integer, primary_key=True),
+ Column('user_id',Integer, ForeignKey('users.id')),
+ autoload=True)
+
+ assert len(a2.foreign_keys) == 1
+ assert len(a2.c.user_id.foreign_keys) == 1
+ assert len(a2.constraints) == 2
+ assert [c.parent for c in a2.foreign_keys] == [a2.c.user_id]
+ assert [c.parent for c in a2.c.user_id.foreign_keys] == [a2.c.user_id]
+ assert list(a2.c.user_id.foreign_keys)[0].parent is a2.c.user_id
+ assert u2.join(a2).onclause == u2.c.id==a2.c.user_id
+
+
finally:
meta.drop_all()
sess.clear()
assert sess.query(User).get(7).name == 'somenewname'
+ def test_no_session(self):
+ mapper(User, users)
+ sess = create_session()
+ u = sess.query(User).get(7)
+
+ sess.expire(u, attribute_names=['name'])
+ sess.expunge(u)
+ try:
+ u.name
+ except exceptions.InvalidRequestError, e:
+ assert str(e) == "Instance <class 'testlib.fixtures.User'> is not bound to a Session, and no contextual session is established; attribute refresh operation cannot proceed"
+
def test_expire_preserves_changes(self):
"""test that the expire load operation doesn't revert post-expire changes"""