Psycopg2 context manager demo: contextmanagers for closing and committing don't play well together without autocommit

close_commit_test.py

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))