Hey all, I'm writing some automated tests for some legacy python code using a psycopg2 connection. I'd like to check data in the database by using SQLAlchemy. I rollback the database transaction after each test in tearDown().
The problem is my SQLAlchemy connection doesn't see the database updates. At first I thought they weren't using the same connection, but they are. I'm using create_engine(..., creator=get_conn). The problem appears to be that DefaultDialect is rolling back the transaction. See sqlalchemy/engine/default.py line 167. I had to review PG logs to spot this as it's not logged. self.do_rollback(connection.connection) Is this line really needed? What would it be rolling back? Can it be avoided? When I disable this line of code, the transaction continues and sqlalchemy can see the updates from psyopg2. I've attached a demo file. Thanks, Brian -- 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.
import os import logging from dotenv import load_dotenv import psycopg2 from psycopg2.extras import LoggingConnection from sqlalchemy import Column, create_engine, Integer, MetaData, Text from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base dotenv_path = os.path.join(os.path.dirname(__file__), '.env') load_dotenv(dotenv_path) logger = logging.getLogger(__name__) logging.basicConfig(level=logging.DEBUG) db_conn = None def get_conn(): global db_conn logging.debug('get_conn()') if db_conn is None: logging.debug(' theres no db_conn, must create one') conn_string = "dbname='{0}' host='localhost' port=5432 user='{1}' password='{2}'".format( os.getenv("DB_NAME"), os.getenv("DB_USER"), os.getenv("DB_PASSWORD")) db_conn = psycopg2.connect(conn_string, connection_factory=LoggingConnection) db_conn.initialize(logger) else: logging.debug(' using exising db_conn') return db_conn BaseCi = declarative_base(metadata=MetaData(schema='config')) class User(BaseCi): __tablename__ = 'user' user_id = Column(Integer, primary_key=True) user_status = Column(Text) def legacy_psycopg_code(uid, status): conn = get_conn() cur = conn.cursor() query = 'UPDATE config.user SET user_status = %s WHERE user_id = %s ' cur.execute(query, (status, uid,)) # Confirm it's there. query = 'select user_status from config.user where user_id = %s' cur.execute(query, (uid,)) uc = cur.fetchone()[0] logging.debug("user_status is %s" % uc) def test_transaction(): uid = 78 status = 'ex-parrot' legacy_psycopg_code(uid, status) # Use existing connection so we can see inside the transaction. engine = create_engine('postgresql+psycopg2://', creator=get_conn) Session = sessionmaker(bind=engine) session = Session() user = session.query(User).get(uid) logging.debug(user.user_status) # assertEquals(user.user_status, status) if __name__ == "__main__": test_transaction()
DEBUG:root:get_conn() DEBUG:root: theres no db_conn, must create one DEBUG:__main__:UPDATE config.user SET user_status = 'ex-parrot' WHERE user_id = 78 DEBUG:__main__:select user_status from config.user where user_id = 78 DEBUG:root:user_status is ex-parrot DEBUG:root:get_conn() DEBUG:root: using exising db_conn DEBUG:__main__:select version() DEBUG:__main__:select current_schema() DEBUG:__main__:show transaction isolation level DEBUG:__main__:SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1 DEBUG:__main__:SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1 DEBUG:__main__:SELECT 'x' AS some_label DEBUG:root:DefaultDialect::initialize() calls do_rollback() here. <----------------------------- DEBUG:__main__:show standard_conforming_strings DEBUG:__main__:SELECT config."user".user_id AS config_user_user_id, config."user".user_status AS config_user_user_status FROM config."user" WHERE config."user".user_id = 78 DEBUG:root:Active