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 sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. Visit this group at https://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.