1import sqlalchemy as sa 2from sqlalchemy import ForeignKey 3from sqlalchemy import Integer 4from sqlalchemy import String 5from sqlalchemy import testing 6from sqlalchemy.orm import mapper 7from sqlalchemy.orm import query 8from sqlalchemy.orm import relationship 9from sqlalchemy.orm import scoped_session 10from sqlalchemy.testing import assert_raises_message 11from sqlalchemy.testing import eq_ 12from sqlalchemy.testing import fixtures 13from sqlalchemy.testing.mock import Mock 14from sqlalchemy.testing.schema import Column 15from sqlalchemy.testing.schema import Table 16 17 18class ScopedSessionTest(fixtures.MappedTest): 19 @classmethod 20 def define_tables(cls, metadata): 21 Table( 22 "table1", 23 metadata, 24 Column( 25 "id", Integer, primary_key=True, test_needs_autoincrement=True 26 ), 27 Column("data", String(30)), 28 ) 29 Table( 30 "table2", 31 metadata, 32 Column( 33 "id", Integer, primary_key=True, test_needs_autoincrement=True 34 ), 35 Column("someid", None, ForeignKey("table1.id")), 36 ) 37 38 def test_basic(self): 39 table2, table1 = self.tables.table2, self.tables.table1 40 41 Session = scoped_session(sa.orm.sessionmaker()) 42 43 class CustomQuery(query.Query): 44 pass 45 46 class SomeObject(fixtures.ComparableEntity): 47 query = Session.query_property() 48 49 class SomeOtherObject(fixtures.ComparableEntity): 50 query = Session.query_property() 51 custom_query = Session.query_property(query_cls=CustomQuery) 52 53 mapper( 54 SomeObject, 55 table1, 56 properties={"options": relationship(SomeOtherObject)}, 57 ) 58 mapper(SomeOtherObject, table2) 59 60 s = SomeObject(id=1, data="hello") 61 sso = SomeOtherObject() 62 s.options.append(sso) 63 Session.add(s) 64 Session.commit() 65 Session.refresh(sso) 66 Session.remove() 67 68 eq_( 69 SomeObject( 70 id=1, data="hello", options=[SomeOtherObject(someid=1)] 71 ), 72 Session.query(SomeObject).one(), 73 ) 74 eq_( 75 SomeObject( 76 id=1, data="hello", options=[SomeOtherObject(someid=1)] 77 ), 78 SomeObject.query.one(), 79 ) 80 eq_( 81 SomeOtherObject(someid=1), 82 SomeOtherObject.query.filter( 83 SomeOtherObject.someid == sso.someid 84 ).one(), 85 ) 86 assert isinstance(SomeOtherObject.query, query.Query) 87 assert not isinstance(SomeOtherObject.query, CustomQuery) 88 assert isinstance(SomeOtherObject.custom_query, query.Query) 89 90 def test_config_errors(self): 91 Session = scoped_session(sa.orm.sessionmaker()) 92 93 s = Session() 94 assert_raises_message( 95 sa.exc.InvalidRequestError, 96 "Scoped session is already present", 97 Session, 98 bind=testing.db, 99 ) 100 101 assert_raises_message( 102 sa.exc.SAWarning, 103 "At least one scoped session is already present. ", 104 Session.configure, 105 bind=testing.db, 106 ) 107 108 def test_call_with_kwargs(self): 109 mock_scope_func = Mock() 110 SessionMaker = sa.orm.sessionmaker() 111 Session = scoped_session(sa.orm.sessionmaker(), mock_scope_func) 112 113 s0 = SessionMaker() 114 assert s0.autocommit == False 115 116 mock_scope_func.return_value = 0 117 s1 = Session() 118 assert s1.autocommit == False 119 120 assert_raises_message( 121 sa.exc.InvalidRequestError, 122 "Scoped session is already present", 123 Session, 124 autocommit=True, 125 ) 126 127 mock_scope_func.return_value = 1 128 s2 = Session(autocommit=True) 129 assert s2.autocommit == True 130