1from sqlalchemy.testing import assert_raises, assert_raises_message 2import sqlalchemy as sa 3from sqlalchemy import testing 4from sqlalchemy.orm import scoped_session 5from sqlalchemy import Integer, String, ForeignKey 6from sqlalchemy.testing.schema import Table, Column 7from sqlalchemy.orm import mapper, relationship, query 8from sqlalchemy.testing import eq_ 9from sqlalchemy.testing import fixtures 10 11 12 13class _ScopedTest(fixtures.MappedTest): 14 """Adds another lookup bucket to emulate Session globals.""" 15 16 run_setup_mappers = 'once' 17 18 @classmethod 19 def setup_class(cls): 20 cls.scoping = _base.adict() 21 super(_ScopedTest, cls).setup_class() 22 23 @classmethod 24 def teardown_class(cls): 25 cls.scoping.clear() 26 super(_ScopedTest, cls).teardown_class() 27 28 29class ScopedSessionTest(fixtures.MappedTest): 30 31 @classmethod 32 def define_tables(cls, metadata): 33 Table('table1', metadata, 34 Column('id', Integer, primary_key=True, test_needs_autoincrement=True), 35 Column('data', String(30))) 36 Table('table2', metadata, 37 Column('id', Integer, primary_key=True, test_needs_autoincrement=True), 38 Column('someid', None, ForeignKey('table1.id'))) 39 40 def test_basic(self): 41 table2, table1 = self.tables.table2, self.tables.table1 42 43 Session = scoped_session(sa.orm.sessionmaker()) 44 45 class CustomQuery(query.Query): 46 pass 47 48 class SomeObject(fixtures.ComparableEntity): 49 query = Session.query_property() 50 class SomeOtherObject(fixtures.ComparableEntity): 51 query = Session.query_property() 52 custom_query = Session.query_property(query_cls=CustomQuery) 53 54 mapper(SomeObject, table1, properties={ 55 'options':relationship(SomeOtherObject)}) 56 mapper(SomeOtherObject, table2) 57 58 s = SomeObject(id=1, data="hello") 59 sso = SomeOtherObject() 60 s.options.append(sso) 61 Session.add(s) 62 Session.commit() 63 Session.refresh(sso) 64 Session.remove() 65 66 eq_(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]), 67 Session.query(SomeObject).one()) 68 eq_(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]), 69 SomeObject.query.one()) 70 eq_(SomeOtherObject(someid=1), 71 SomeOtherObject.query.filter( 72 SomeOtherObject.someid == sso.someid).one()) 73 assert isinstance(SomeOtherObject.query, query.Query) 74 assert not isinstance(SomeOtherObject.query, CustomQuery) 75 assert isinstance(SomeOtherObject.custom_query, query.Query) 76 77 def test_config_errors(self): 78 Session = scoped_session(sa.orm.sessionmaker()) 79 80 s = Session() 81 assert_raises_message( 82 sa.exc.InvalidRequestError, 83 "Scoped session is already present", 84 Session, bind=testing.db 85 ) 86 87 assert_raises_message( 88 sa.exc.SAWarning, 89 "At least one scoped session is already present. ", 90 Session.configure, bind=testing.db 91 ) 92 93 94 95