Mixing
SQLAlchemy session management in test-agnostic functions
up vote
1
down vote
favorite
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 5 hours ago
Yuval Adam
108k75263357
add a comment |
up vote
1
down vote
favorite
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 5 hours ago
Yuval Adam
108k75263357
Is monkeypatchingget_sessionto use thesessionfixture returns an option?
– hoefling
5 hours ago
@hoefling sure, do you have an example for that?
– Yuval Adam
5 hours ago
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ä
3 hours ago
@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
3 hours ago
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ä
3 hours ago
add a comment |
up vote
1
down vote
favorite
up vote
1
down vote
favorite
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 5 hours ago
Yuval Adam
108k75263357
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 5 hours ago
Yuval Adam
108k75263357
asked 5 hours ago
Yuval Adam
108k75263357
asked 5 hours ago
Yuval Adam
108k75263357
asked 5 hours ago
Yuval Adam
108k75263357
asked 5 hours ago
Yuval Adam
108k75263357
108k75263357
Is monkeypatchingget_sessionto use thesessionfixture returns an option?
– hoefling
5 hours ago
@hoefling sure, do you have an example for that?
– Yuval Adam
5 hours ago
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ä
3 hours ago
@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
3 hours ago
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ä
3 hours ago
add a comment |
Is monkeypatchingget_sessionto use thesessionfixture returns an option?
– hoefling
5 hours ago
@hoefling sure, do you have an example for that?
– Yuval Adam
5 hours ago
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ä
3 hours ago
@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
3 hours ago
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ä
3 hours ago
Is monkeypatching
get_session to use the session fixture returns an option?– hoefling
5 hours ago
Is monkeypatching
get_session to use the session fixture returns an option?– hoefling
5 hours ago
@hoefling sure, do you have an example for that?
– Yuval Adam
5 hours ago
@hoefling sure, do you have an example for that?
– Yuval Adam
5 hours ago
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ä
3 hours ago
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ä
3 hours ago
@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
3 hours ago
@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
3 hours ago
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ä
3 hours ago
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ä
3 hours ago
add a comment |
active
oldest
votes
active
oldest
votes
active
oldest
votes
active
oldest
votes
active
oldest
votes
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
5 hours ago
@hoefling sure, do you have an example for that?
– Yuval Adam
5 hours ago
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ä
3 hours ago
@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
3 hours ago
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ä
3 hours ago