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?










share|improve this question






















  • Is monkeypatching get_session to use the session fixture returns an option?
    – hoefling
    yesterday










  • @hoefling sure, do you have an example for that?
    – Yuval Adam
    yesterday






  • 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ä
    yesterday












  • @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
    yesterday








  • 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ä
    yesterday

















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?










share|improve this question






















  • Is monkeypatching get_session to use the session fixture returns an option?
    – hoefling
    yesterday










  • @hoefling sure, do you have an example for that?
    – Yuval Adam
    yesterday






  • 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ä
    yesterday












  • @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
    yesterday








  • 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ä
    yesterday















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?










share|improve this question













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






share|improve this question













share|improve this question











share|improve this question




share|improve this question










asked yesterday









Yuval Adam

108k75263358




108k75263358












  • Is monkeypatching get_session to use the session fixture returns an option?
    – hoefling
    yesterday










  • @hoefling sure, do you have an example for that?
    – Yuval Adam
    yesterday






  • 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ä
    yesterday












  • @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
    yesterday








  • 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ä
    yesterday




















  • Is monkeypatching get_session to use the session fixture returns an option?
    – hoefling
    yesterday










  • @hoefling sure, do you have an example for that?
    – Yuval Adam
    yesterday






  • 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ä
    yesterday












  • @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
    yesterday








  • 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ä
    yesterday


















Is monkeypatching get_session to use the session fixture returns an option?
– hoefling
yesterday




Is monkeypatching get_session to use the session fixture returns an option?
– hoefling
yesterday












@hoefling sure, do you have an example for that?
– Yuval Adam
yesterday




@hoefling sure, do you have an example for that?
– Yuval Adam
yesterday




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ä
yesterday






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ä
yesterday














@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
yesterday






@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
yesterday






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ä
yesterday






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ä
yesterday



















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',
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
});


}
});














 

draft saved


draft discarded


















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






























active

oldest

votes













active

oldest

votes









active

oldest

votes






active

oldest

votes
















 

draft saved


draft discarded



















































 


draft saved


draft discarded














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





















































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







Popular posts from this blog

404 Error Contact Form 7 ajax form submitting

How to know if a Active Directory user can login interactively

TypeError: fit_transform() missing 1 required positional argument: 'X'