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 [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.
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
