I'm trying to execute a fairly simple UPDATE query.
query = update(Model).where(Model.id.in_(list_of_ids)).values(x=1)
I know of two methods to execute it. One using the session and the other
using the engine. However, depending on which I use, the results I get are
very different.
db.session.execute(query) # works in test suite but not live.
db.engine.connect().execute(query) # works live but not in test suite.
I'm trying to understand why this would be the case. I believe either my
test suite of my implementation of sqlalchemy is broken. I was wondering
if you had any tips.
Test Suite:
def setUp(self):
"""Create app test client."""
self.app = app
self.app_context = self.app.app_context()
self.app_context.push()
self.client = self.app.test_client()
self.transaction = connection.begin()
db.session = scoped_session(
sessionmaker(bind=connection, query_cls=db.query_class))
db.session.begin_nested()
@event.listens_for(db.session, "after_transaction_end")
def restart_savepoint(session, transaction):
if transaction.nested and not transaction._parent.nested:
session.expire_all()
session.begin_nested()
# this is a cleanup function rather than a teardown function in case
# the db gets into a bad state and setup fails, in which case we still
# want the drop_all to be called
self.addCleanup(self.cleanup)
def cleanup(self):
"""Tear down database."""
db.session.close()
self.transaction.rollback()
self.app_context.pop()
@classmethod
def setUpClass(cls):
"""Create the database."""
global app, engine, connection
app = cls._create_app()
engine = db.engine
connection = engine.connect()
@classmethod
def tearDownClass(cls):
"""Destroy the database."""
connection.close()
engine.dispose()
Session construction when live:
@property
def engine(self):
"""Return an engine instance."""
if not self._engine:
database_uri = self.app.config.get('SQLALCHEMY_DATABASE_URI')
self._engine = sqlalchemy.create_engine(database_uri)
return self._engine
@property
def session(self):
"""Return database session."""
if not self._session:
factory = sqlalchemy.orm.sessionmaker(
bind=self.engine, query_cls=self.query_class)
self._session = sqlalchemy.orm.scoped_session(
factory, scopefunc=_app_ctx_stack.__ident_func__)
return self._session()
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
