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