SQLAlchemy session management in test-agnostic functions
I'm working on a Python application that uses the following SQLAlchemy pattern for separating unit tests into isolated transactions, and using pytest fixtures:
@fixture(scope='function')
def session(engine):
connection = engine.connect()
transaction = connection.begin()
Session = get_session(bind=connection)
session = Session()
session.begin_nested()
@listens_for(session, 'after_transaction_end')
def resetart_savepoint(sess, trans):
if trans.nested and not trans._parent.nested:
session.expire_all()
session.begin_nested()
yield session
session.close()
transaction.rollback()
connection.close()
I'm defining the get_session() utility function such that I can also use it in my application code:
def get_session(bind=None):
if bind is None:
bind = create_engine(DATABASE_URL)
return scoped_session(sessionmaker(bind=bind))
This pattern works flawlessly in unit test code that uses the session fixture directly:
def test_foo(session):
assert session.query(Foo).count() == 0
However, whenever I want to test application code that should access the session by itself, without it being passed explicitly from the unit test, I get a failure:
def create_foo_obj():
Session = get_session()
session = Session()
session.add(Foo())
session.commit()
def test_foo(session):
create_foo_obj()
assert session.query(Foo).count() == 1 # FAILS
Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session that is bound from the test fixture, but that doesn't happen.
I know I can workaround this by passing the session directly to the function, but that seems just wrong.
Any better way to accomplish this?
python python-3.x sqlalchemy pytest
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
add a comment |
I'm working on a Python application that uses the following SQLAlchemy pattern for separating unit tests into isolated transactions, and using pytest fixtures:
@fixture(scope='function')
def session(engine):
connection = engine.connect()
transaction = connection.begin()
Session = get_session(bind=connection)
session = Session()
session.begin_nested()
@listens_for(session, 'after_transaction_end')
def resetart_savepoint(sess, trans):
if trans.nested and not trans._parent.nested:
session.expire_all()
session.begin_nested()
yield session
session.close()
transaction.rollback()
connection.close()
I'm defining the get_session() utility function such that I can also use it in my application code:
def get_session(bind=None):
if bind is None:
bind = create_engine(DATABASE_URL)
return scoped_session(sessionmaker(bind=bind))
This pattern works flawlessly in unit test code that uses the session fixture directly:
def test_foo(session):
assert session.query(Foo).count() == 0
However, whenever I want to test application code that should access the session by itself, without it being passed explicitly from the unit test, I get a failure:
def create_foo_obj():
Session = get_session()
session = Session()
session.add(Foo())
session.commit()
def test_foo(session):
create_foo_obj()
assert session.query(Foo).count() == 1 # FAILS
Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session that is bound from the test fixture, but that doesn't happen.
I know I can workaround this by passing the session directly to the function, but that seems just wrong.
Any better way to accomplish this?
python python-3.x sqlalchemy pytest
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
Is monkeypatchingget_sessionto use thesessionfixture returns an option?
– hoefling
Nov 19 '18 at 9:31
@hoefling sure, do you have an example for that?
– Yuval Adam
Nov 19 '18 at 9:43
2
Your expectations would seem to be a bit off regarding "Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session". The function seems to create a new scoped session (and session factory) per call. Also, passing a session to functions that need it is right in most cases.
– Ilja Everilä
Nov 19 '18 at 11:09
@IljaEverilä my use case is testing Celery tasks. How would you manage sessions for those tasks? Currently my thought is to have a top levelSessionand createsession = Session()within each task. Passing a session parameter to a celery task doesn't really make sense (it's probably not serializable). Is there a better way you can suggest?
– Yuval Adam
Nov 19 '18 at 11:35
1
Disclaimer: haven't used Celery ever, but sounds like you're on the right track in that a task is a natural lifetime for aSession, given that they're not too long lived. That way a tasks DB operations all succeed or fail as one etc., if you handle the transaction at the task level and the functions etc. used by the task just receive the session from the task. If possible, just make it so that theSessionused by a task is injectable so that you can slip in a mock when necessary, or a test controlled one etc.
– Ilja Everilä
Nov 19 '18 at 11:48
add a comment |
I'm working on a Python application that uses the following SQLAlchemy pattern for separating unit tests into isolated transactions, and using pytest fixtures:
@fixture(scope='function')
def session(engine):
connection = engine.connect()
transaction = connection.begin()
Session = get_session(bind=connection)
session = Session()
session.begin_nested()
@listens_for(session, 'after_transaction_end')
def resetart_savepoint(sess, trans):
if trans.nested and not trans._parent.nested:
session.expire_all()
session.begin_nested()
yield session
session.close()
transaction.rollback()
connection.close()
I'm defining the get_session() utility function such that I can also use it in my application code:
def get_session(bind=None):
if bind is None:
bind = create_engine(DATABASE_URL)
return scoped_session(sessionmaker(bind=bind))
This pattern works flawlessly in unit test code that uses the session fixture directly:
def test_foo(session):
assert session.query(Foo).count() == 0
However, whenever I want to test application code that should access the session by itself, without it being passed explicitly from the unit test, I get a failure:
def create_foo_obj():
Session = get_session()
session = Session()
session.add(Foo())
session.commit()
def test_foo(session):
create_foo_obj()
assert session.query(Foo).count() == 1 # FAILS
Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session that is bound from the test fixture, but that doesn't happen.
I know I can workaround this by passing the session directly to the function, but that seems just wrong.
Any better way to accomplish this?
python python-3.x sqlalchemy pytest
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
I'm working on a Python application that uses the following SQLAlchemy pattern for separating unit tests into isolated transactions, and using pytest fixtures:
@fixture(scope='function')
def session(engine):
connection = engine.connect()
transaction = connection.begin()
Session = get_session(bind=connection)
session = Session()
session.begin_nested()
@listens_for(session, 'after_transaction_end')
def resetart_savepoint(sess, trans):
if trans.nested and not trans._parent.nested:
session.expire_all()
session.begin_nested()
yield session
session.close()
transaction.rollback()
connection.close()
I'm defining the get_session() utility function such that I can also use it in my application code:
def get_session(bind=None):
if bind is None:
bind = create_engine(DATABASE_URL)
return scoped_session(sessionmaker(bind=bind))
This pattern works flawlessly in unit test code that uses the session fixture directly:
def test_foo(session):
assert session.query(Foo).count() == 0
However, whenever I want to test application code that should access the session by itself, without it being passed explicitly from the unit test, I get a failure:
def create_foo_obj():
Session = get_session()
session = Session()
session.add(Foo())
session.commit()
def test_foo(session):
create_foo_obj()
assert session.query(Foo).count() == 1 # FAILS
Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session that is bound from the test fixture, but that doesn't happen.
I know I can workaround this by passing the session directly to the function, but that seems just wrong.
Any better way to accomplish this?
python python-3.x sqlalchemy pytest
python python-3.x sqlalchemy pytest
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
asked Nov 19 '18 at 9:13
Yuval AdamYuval Adam
110k75264361
110k75264361
Is monkeypatchingget_sessionto use thesessionfixture returns an option?
– hoefling
Nov 19 '18 at 9:31
@hoefling sure, do you have an example for that?
– Yuval Adam
Nov 19 '18 at 9:43
2
Your expectations would seem to be a bit off regarding "Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session". The function seems to create a new scoped session (and session factory) per call. Also, passing a session to functions that need it is right in most cases.
– Ilja Everilä
Nov 19 '18 at 11:09
@IljaEverilä my use case is testing Celery tasks. How would you manage sessions for those tasks? Currently my thought is to have a top levelSessionand createsession = Session()within each task. Passing a session parameter to a celery task doesn't really make sense (it's probably not serializable). Is there a better way you can suggest?
– Yuval Adam
Nov 19 '18 at 11:35
1
Disclaimer: haven't used Celery ever, but sounds like you're on the right track in that a task is a natural lifetime for aSession, given that they're not too long lived. That way a tasks DB operations all succeed or fail as one etc., if you handle the transaction at the task level and the functions etc. used by the task just receive the session from the task. If possible, just make it so that theSessionused by a task is injectable so that you can slip in a mock when necessary, or a test controlled one etc.
– Ilja Everilä
Nov 19 '18 at 11:48
add a comment |
Is monkeypatchingget_sessionto use thesessionfixture returns an option?
– hoefling
Nov 19 '18 at 9:31
@hoefling sure, do you have an example for that?
– Yuval Adam
Nov 19 '18 at 9:43
2
Your expectations would seem to be a bit off regarding "Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session". The function seems to create a new scoped session (and session factory) per call. Also, passing a session to functions that need it is right in most cases.
– Ilja Everilä
Nov 19 '18 at 11:09
@IljaEverilä my use case is testing Celery tasks. How would you manage sessions for those tasks? Currently my thought is to have a top levelSessionand createsession = Session()within each task. Passing a session parameter to a celery task doesn't really make sense (it's probably not serializable). Is there a better way you can suggest?
– Yuval Adam
Nov 19 '18 at 11:35
1
Disclaimer: haven't used Celery ever, but sounds like you're on the right track in that a task is a natural lifetime for aSession, given that they're not too long lived. That way a tasks DB operations all succeed or fail as one etc., if you handle the transaction at the task level and the functions etc. used by the task just receive the session from the task. If possible, just make it so that theSessionused by a task is injectable so that you can slip in a mock when necessary, or a test controlled one etc.
– Ilja Everilä
Nov 19 '18 at 11:48
Is monkeypatching
get_session to use the session fixture returns an option?– hoefling
Nov 19 '18 at 9:31
Is monkeypatching
get_session to use the session fixture returns an option?– hoefling
Nov 19 '18 at 9:31
@hoefling sure, do you have an example for that?
– Yuval Adam
Nov 19 '18 at 9:43
@hoefling sure, do you have an example for that?
– Yuval Adam
Nov 19 '18 at 9:43
2
2
Your expectations would seem to be a bit off regarding "Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session". The function seems to create a new scoped session (and session factory) per call. Also, passing a session to functions that need it is right in most cases.
– Ilja Everilä
Nov 19 '18 at 11:09
Your expectations would seem to be a bit off regarding "Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session". The function seems to create a new scoped session (and session factory) per call. Also, passing a session to functions that need it is right in most cases.
– Ilja Everilä
Nov 19 '18 at 11:09
@IljaEverilä my use case is testing Celery tasks. How would you manage sessions for those tasks? Currently my thought is to have a top level
Session and create session = Session() within each task. Passing a session parameter to a celery task doesn't really make sense (it's probably not serializable). Is there a better way you can suggest?– Yuval Adam
Nov 19 '18 at 11:35
@IljaEverilä my use case is testing Celery tasks. How would you manage sessions for those tasks? Currently my thought is to have a top level
Session and create session = Session() within each task. Passing a session parameter to a celery task doesn't really make sense (it's probably not serializable). Is there a better way you can suggest?– Yuval Adam
Nov 19 '18 at 11:35
1
1
Disclaimer: haven't used Celery ever, but sounds like you're on the right track in that a task is a natural lifetime for a
Session, given that they're not too long lived. That way a tasks DB operations all succeed or fail as one etc., if you handle the transaction at the task level and the functions etc. used by the task just receive the session from the task. If possible, just make it so that the Session used by a task is injectable so that you can slip in a mock when necessary, or a test controlled one etc.– Ilja Everilä
Nov 19 '18 at 11:48
Disclaimer: haven't used Celery ever, but sounds like you're on the right track in that a task is a natural lifetime for a
Session, given that they're not too long lived. That way a tasks DB operations all succeed or fail as one etc., if you handle the transaction at the task level and the functions etc. used by the task just receive the session from the task. If possible, just make it so that the Session used by a task is injectable so that you can slip in a mock when necessary, or a test controlled one etc.– Ilja Everilä
Nov 19 '18 at 11:48
add a comment |
0
active
oldest
votes
Your Answer
StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");
StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);
StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});
function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});
}
});
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53371410%2fsqlalchemy-session-management-in-test-agnostic-functions%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
0
active
oldest
votes
0
active
oldest
votes
active
oldest
votes
active
oldest
votes
Thanks for contributing an answer to Stack Overflow!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53371410%2fsqlalchemy-session-management-in-test-agnostic-functions%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Is monkeypatching
get_sessionto use thesessionfixture returns an option?– hoefling
Nov 19 '18 at 9:31
@hoefling sure, do you have an example for that?
– Yuval Adam
Nov 19 '18 at 9:43
2
Your expectations would seem to be a bit off regarding "Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session". The function seems to create a new scoped session (and session factory) per call. Also, passing a session to functions that need it is right in most cases.
– Ilja Everilä
Nov 19 '18 at 11:09
@IljaEverilä my use case is testing Celery tasks. How would you manage sessions for those tasks? Currently my thought is to have a top level
Sessionand createsession = Session()within each task. Passing a session parameter to a celery task doesn't really make sense (it's probably not serializable). Is there a better way you can suggest?– Yuval Adam
Nov 19 '18 at 11:35
1
Disclaimer: haven't used Celery ever, but sounds like you're on the right track in that a task is a natural lifetime for a
Session, given that they're not too long lived. That way a tasks DB operations all succeed or fail as one etc., if you handle the transaction at the task level and the functions etc. used by the task just receive the session from the task. If possible, just make it so that theSessionused by a task is injectable so that you can slip in a mock when necessary, or a test controlled one etc.– Ilja Everilä
Nov 19 '18 at 11:48