Skip to content

Instantly share code, notes, and snippets.

@bhtucker
Last active February 24, 2022 20:00
Show Gist options
  • Save bhtucker/7c63881dbf520e426975b1e5f9b6832c to your computer and use it in GitHub Desktop.
Save bhtucker/7c63881dbf520e426975b1e5f9b6832c to your computer and use it in GitHub Desktop.
Psycopg2 context manager demo: contextmanagers for closing and committing don't play well together without autocommit
import psycopg2
from contextlib import closing
"""
Demonstrate that contextmanagers for closing and committing don't play well together without autocommit
Only the noclosing function can run without error
"""
def withclosing():
with closing(psycopg2.connect(database='bensontucker')) as conn, conn.cursor() as cur:
cur.execute('create table if not exists test_close (id int)')
with closing(psycopg2.connect(database='bensontucker')) as conn, conn.cursor() as cur:
cur.execute('insert into test_close (id) values (1)')
with closing(psycopg2.connect(database='bensontucker')) as conn, conn.cursor() as cur:
cur.execute('select * from test_close')
print(cur.fetchall())
print('connection is ' + ('closed' if conn.closed else 'open'))
def with_nested_closing():
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
with closing(conn) as conn:
cur.execute('create table if not exists test_nested_closing (id int)')
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
with closing(conn) as conn:
cur.execute('insert into test_nested_closing (id) values (1)')
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
with closing(conn) as conn:
cur.execute('select * from test_nested_closing')
print(cur.fetchall())
print('connection is ' + ('closed' if conn.closed else 'open'))
def noclosing():
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
cur.execute('create table if not exists test_noclose (id int)')
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
cur.execute('insert into test_noclose (id) values (1)')
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
cur.execute('select * from test_noclose')
print(cur.fetchall())
print('connection is ' + ('closed' if conn.closed else 'open'))
def finallyclose():
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
try:
cur.execute('create table if not exists finallyclose (id int)')
finally:
conn.close()
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
try:
cur.execute('insert into finallyclose (id) values (1)')
finally:
conn.close()
with psycopg2.connect(database='bensontucker') as conn, conn.cursor() as cur:
try:
cur.execute('select * from finallyclose')
print(cur.fetchall())
finally:
conn.close()
print('connection is ' + ('closed' if conn.closed else 'open'))
if __name__ == '__main__':
print('Trying without closing wrapper')
noclosing()
print('Trying with finally:')
try:
finallyclose()
except Exception as e:
print('Exception:', str(e))
print('Trying with closing wrapper')
try:
withclosing()
except Exception as e:
print('Exception:', str(e))
print('Trying with nested closing:')
try:
with_nested_closing()
except Exception as e:
print('Exception:', str(e))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment