Skip to content

Instantly share code, notes, and snippets.

@charlax
Created September 7, 2012 08:35
Show Gist options
  • Save charlax/3664378 to your computer and use it in GitHub Desktop.
Save charlax/3664378 to your computer and use it in GitHub Desktop.
Transaction and scopedsession (SQLAlchemy & Pyramid)
from sqlalchemy.orm import sessionmaker, scoped_session
...
Session = scoped_session(sessionmaker())
# When using scoped_session, Session and Session() expose the same .commit() .query() .add() etc. methods.
...
from myapp.model import Session, Base
class User(Base):
...
# User model is using the myapp.model's Session to perform pre-insert and pre-update
# And there are some other methods like:
@classmethod
def get_by_name(cls, name):
return Session.query(...)
# And some other that might write stuff:
def set_name(self, name):
self.name = name
Session.add(self)
Session.commit()
# Printing the Session here gives
# <sqlalchemy.orm.scoping.ScopedSession object at 0x1038b95d0>
# When running the tests, will give me:
# InvalidRequestError: Object '<User at 0x104ee5350>' is already attached to session '1' (this is '2')
pass
# From http://sontek.net/blog/detail/writing-tests-for-pyramid-and-sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class MyAppTest(unittest.TestCase):
@classmethod
def setup_class(cls):
cls.engine = create_engine(
'mysql://...@127.0.0.1:3306/myapptest?charset=utf8&use_unicode=0',
echo = False,
)
cls.Session = sessionmaker()
def setUp(self):
connection = self.engine.connect()
# begin a non-ORM transaction
self.trans = connection.begin()
# bind an individual Session to the connection
self.session = self.Session(bind=connection)
# HERE'S THE QUESTION
# Since the model also writes stuff to the db (see dummy function User.set_name below), I also need
# to make sure Session is wrapped in a Session, do I?
# Here is my naive tentative
from myapp import model
# This line gives "InvalidRequestError: Scoped session is already present; no new arguments may be specified."
# when the second test is run
model.Session(bind=connection)
print "in test.setUp: self.session: %s" % self.session
# Prints <sqlalchemy.orm.session.Session object at 0x106fdee90>
print "in test.setUp: model.Session: %s" % model.Session
# Prints <sqlalchemy.orm.scoping.ScopedSession object at 0x105f195d0>
from myapp.model import Session
print "in test.setUp: Session: %s" % Session
# Prints <sqlalchemy.orm.scoping.ScopedSession object at 0x105f195d0>
from myapp.tests import MyAppTest
from myapp.model.user import User
class TestWhatever(MyAppTest):
user = User(name="Charles")
def setUp(self):
# Installing some fixtures
self.session.add(self.user)
self.session.flush()
def test_set_name(self):
# Just as an example, obviously not an actual test
self.user.set_name("Maxwell")
self.assertEqual(self.user.name, "Maxwell")
def test_set_name2(self):
# Just as an example, obviously not an actual test
self.user.set_name("John")
self.assertEqual(self.user.name, "John")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment