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