Transactions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Transactions with [`psycopg`](https://www.psycopg.org/psycopg3/docs/index.html)\n",
    "\n",
    "In this notebook, we showcase how to connect to a [PostgreSQL](https://www.postgresql.org/) database, execute queries, and run transactions with different isolation levels using the [`psycopg`](https://www.psycopg.org/psycopg3/docs/index.html) package for python. The notebook is inspired by [this](https://pynative.com/python-postgresql-transaction-management-using-commit-and-rollback/) website.\n",
    "\n",
    "Copyright Marcel Maltry & Jens Dittrich, [Big Data Analytics Group](https://bigdata.uni-saarland.de/), [CC-BY-SA](https://creativecommons.org/licenses/by-sa/4.0/legalcode)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup\n",
    "\n",
    "The following cell serves as setup. We will explain the syntax in more detail below. Here, we simply connect to the database, create a new table `accounts` with attributes `id` and `balance`, and add some toy data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import psycopg\n",
    "\n",
    "# dsn depends on installation type\n",
    "dsn = 'dbname=postgres user=postgres host=/var/run/postgresql/'  # vagrant VM, host may be /tmp/ on other systems\n",
    "#dsn = 'dbname=postgres user=postgres host=db' # docker\n",
    "\n",
    "def reset_db(dsn):\n",
    "    # Connect to postgres user's default database\n",
    "    with psycopg.connect(dsn) as conn:\n",
    "\n",
    "        # Open a cursor to perform database operations\n",
    "        with conn.cursor() as cur:\n",
    "\n",
    "            # Drop table if existing\n",
    "            cur.execute(\"\"\"DROP TABLE IF EXISTS accounts;\"\"\")\n",
    "\n",
    "            # Create accounts table\n",
    "            cur.execute(\"\"\"CREATE TABLE accounts\n",
    "                          (id int PRIMARY KEY, balance float(2));\"\"\")\n",
    "\n",
    "            # Insert sample data into accounts table\n",
    "            cur.execute(\"\"\"INSERT INTO accounts\n",
    "                           VALUES\n",
    "                               (1, 2000.0),\n",
    "                               (2, 520.0),\n",
    "                               (3, 470.0),\n",
    "                               (4, 1700.0),\n",
    "                               (5, 2400.0);\"\"\")\n",
    "\n",
    "            # Note: if no exception has been raised by the block, the transaction is commited implicitly.\n",
    "            \n",
    "reset_db(dsn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Basics\n",
    "\n",
    "In order to send queries to the database, we first need to establish a `connection`. We call the `connect()` method and provide some basic connection parameters such as the database name, the user, the password, and the host, if the database we try to connect to is running on a remote.\n",
    "\n",
    "We can send queries to the database with a cursor that has to be opened from an established connection via the `cursor()` method. The cursor allows us both to send queries (`execute()`) and retrieve results (`fetchone()`, `fetchall()`). Results are always tuples, even if they consist of a single integer. We have to consider this when parsing the results. When we are done, we close the cursor and the connection (this happens implicitly using the [`with`](https://docs.python.org/3/reference/compound_stmts.html#the-with-statement) statement.\n",
    "\n",
    "The following example shows how to query the database for an entire table. We also use the cursor to obtain some additional information on the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The query was executed with status message \"SELECT 5\".\n",
      "The query returned 5 rows that can be fetched.\n",
      "The cursor currently points to row 0.\n",
      "The description of the query result is [<Column 'id', type: int4 (oid: 23)>, <Column 'balance', type: float4 (oid: 700)>].\n",
      "The query returned the following tuples:\n",
      "[(1, 2000.0), (2, 520.0), (3, 470.0), (4, 1700.0), (5, 2400.0)]\n"
     ]
    }
   ],
   "source": [
    "# Connect to database as specified in dsn\n",
    "with psycopg.connect(dsn) as conn:\n",
    "    # Open a cursor to perform database operations\n",
    "    with conn.cursor() as cur:\n",
    "        # Define a SQL query\n",
    "        q_accounts = \"\"\"SELECT * FROM accounts;\"\"\"\n",
    "\n",
    "        # Execute the query using the cursor\n",
    "        cur.execute(q_accounts)\n",
    "\n",
    "        # Print information on the query and its result\n",
    "        print(f\"The query was executed with status message \\\"{cur.statusmessage}\\\".\")\n",
    "        print(f\"The query returned {cur.rowcount} rows that can be fetched.\")\n",
    "        print(f\"The cursor currently points to row {cur.rownumber}.\")\n",
    "        print(f\"The description of the query result is {cur.description}.\")\n",
    "\n",
    "        # Fetch results from cursor\n",
    "        accounts = cur.fetchall()\n",
    "\n",
    "        # Print sorted results\n",
    "        print(f\"The query returned the following tuples:\\n{sorted(accounts)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Session Parameters\n",
    "\n",
    "`psycopg` allows us to set certain session parameters that handle, how the next transaction of a connection is executed. In the following, we will take a closer look at:\n",
    "\n",
    "* `read_only`: The session is set to read-only and, thus, write operations will fail with an exception.\n",
    "* `autocommit`: Every statement sent to the database has an immediate effect, i.e. each statement is an individual transaction that is implicitly commited upon successful completion.\n",
    "* `isolation_level`: This allows us to choose from multiple isolation levels.\n",
    "\n",
    "Session parameters can be set explicitly, e.g. `conn.read_only = True`. Note that session parameters of a connection can only be set if no transaction is currently being performed on the connection.\n",
    "\n",
    "\n",
    "## Read-Only\n",
    "\n",
    "If we set a connection to read-only mode, write operations will not be executed and instead will raise an `ReadOnlySqlTransaction` exception. The following example demonstrates this behavior."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ERROR: The query failed due to the connection being read-only.\n"
     ]
    }
   ],
   "source": [
    "with psycopg.connect(dsn) as conn:\n",
    "    conn.read_only = True\n",
    "    with conn.cursor() as cur:\n",
    "        try:\n",
    "            # Try to insert a new tuple into the table\n",
    "            cur.execute(\"INSERT INTO accounts VALUES (6, 100000.0);\")\n",
    "\n",
    "            # If successful, print newly added tuple\n",
    "            cur.execute(\"SELECT * FROM accounts WHERE id=6;\")\n",
    "            print(cur.fetchone())\n",
    "        except psycopg.errors.ReadOnlySqlTransaction:\n",
    "            print(f\"ERROR: The query failed due to the connection being read-only.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Auto-Commit\n",
    "\n",
    "If `autocommit` is set to `True`, each call of `cur.execute()` is handled as an individual transactions and will either have an immediate effect or fail. The following example shows that each modification is immediately visible to other connections to the database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Due to NO auto-commit, the tuple with id=6 is NOT visible to the other connection: []\n"
     ]
    }
   ],
   "source": [
    "# no auto-commit\n",
    "with psycopg.connect(dsn) as conn1:\n",
    "    # Open a second connection to the database in auto-commit mode\n",
    "    with psycopg.connect(dsn) as conn2:  \n",
    "        with conn1.cursor() as cur1:\n",
    "            cur1.execute(\"\"\"INSERT INTO accounts VALUES (6, 237.0);\"\"\") \n",
    "        with conn2.cursor() as cur2:\n",
    "            cur2.execute(\"\"\"SELECT * FROM accounts WHERE id=6;\"\"\")\n",
    "            print(f\"Due to NO auto-commit, the tuple with id=6 is NOT visible to the other connection: {cur2.fetchall()}\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(6, 237.0)]\n"
     ]
    }
   ],
   "source": [
    "# after connection 1 is committed, the tuple is visible\n",
    "with psycopg.connect(dsn) as conn:\n",
    "    with conn.cursor() as cur:\n",
    "        cur.execute(\"\"\"SELECT * FROM accounts WHERE id=6;\"\"\")\n",
    "        print(f'{cur.fetchall()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Due to auto-commit, the tuple with id=7 is already visible to the other connection: [(7, 237.0)]\n"
     ]
    }
   ],
   "source": [
    "# auto-commit, same code as above (including autocommit)\n",
    "with psycopg.connect(dsn) as conn1:\n",
    "    # Set connection to auto-commit mode\n",
    "    conn1.autocommit = True\n",
    "    with psycopg.connect(dsn) as conn2:\n",
    "        # Set connection to auto-commit mode\n",
    "        conn2.autocommit = True        \n",
    "        with conn1.cursor() as cur1:\n",
    "            cur1.execute(\"\"\"INSERT INTO accounts VALUES (7, 237.0);\"\"\") \n",
    "        # Use cursor from second connection to see immediate effect\n",
    "        with conn2.cursor() as cur2:\n",
    "            cur2.execute(\"\"\"SELECT * FROM accounts WHERE id=7;\"\"\")\n",
    "            print(f\"Due to auto-commit, the tuple with id=7 is already visible to the other connection: {cur2.fetchall()}\") "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Commit Transaction\n",
    "\n",
    "If we set `autocommit` to `False` (this is the default setting), the first call of execute on a cursor begins a new transaction ([`BEGIN`](https://www.postgresql.org/docs/current/sql-begin.html)) and `commit()` has to be called explicitly to make the transaction persistent ([`COMMIT`](https://www.postgresql.org/docs/current/sql-commit.html)). The example below transfers money from one account to another. It is equivalent to running the following transaction directly from the database shell:\n",
    "```SQL\n",
    "BEGIN;\n",
    "UPDATE accounts SET balance = balance - 100 WHERE id=3;\n",
    "UPDATE accounts SET balance = balance + 100 WHERE id=1;\n",
    "COMMIT;\n",
    "```\n",
    "We also show that as long as the transaction is not commited, changes are not visible to other connections. Note that this also depends on the isolation level (we will get back to this).\n",
    "\n",
    "### Show visibility of local changes:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Account balances observed by each connection before COMMIT:\n",
      "Transaction 1: [(1, 2100.0), (3, 370.0)]\n",
      "Transaction 2: [(1, 2000.0), (3, 470.0)]\n",
      "Changes not yet visible to connection 2.\n",
      "--Transaction 1 commited--\n",
      "Account balances observed by each connection after COMMIT:\n",
      "Transaction 1: [(1, 2100.0), (3, 370.0)]\n",
      "Transaction 2: [(1, 2100.0), (3, 370.0)]\n",
      "Changes visible to transaction 2.\n"
     ]
    }
   ],
   "source": [
    "with psycopg.connect(dsn) as conn1:\n",
    "    # Set connection to transaction mode\n",
    "    conn1.autocommit = False\n",
    "    with psycopg.connect(dsn) as conn2:\n",
    "        # Set connection to autocommit mode\n",
    "        conn2.autocommit = True        \n",
    "        with conn1.cursor() as cur1:\n",
    "             with conn2.cursor() as cur2:\n",
    "                # Update balance of account 3, implicitly begins a transaction\n",
    "                cur1.execute(\"\"\"UPDATE accounts SET balance = balance - 100 WHERE id=3;\"\"\") \n",
    "                # Update balance of account 1, implicitly begins a transaction\n",
    "                cur1.execute(\"\"\"UPDATE accounts SET balance = balance + 100 WHERE id=1;\"\"\")\n",
    "                # Compare states visible to both transactions\n",
    "                q_acc = \"\"\"SELECT * FROM accounts WHERE id=1 OR id=3;\"\"\"\n",
    "                cur1.execute(q_acc)\n",
    "                cur2.execute(q_acc)\n",
    "                print(f\"Account balances observed by each connection before COMMIT:\\n\"\\\n",
    "                      f\"Transaction 1: {cur1.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2: {cur2.fetchall()}\\n\"\\\n",
    "                      f\"Changes not yet visible to connection 2.\"\\\n",
    "                     )\n",
    "                # Explicitly commit transaction 1\n",
    "                conn1.commit()\n",
    "                print(\"--Transaction 1 commited--\")\n",
    "                # Compare states visible to both transactions\n",
    "                cur1.execute(q_acc)\n",
    "                cur2.execute(q_acc)\n",
    "                print(f\"Account balances observed by each connection after COMMIT:\\n\"\\\n",
    "                      f\"Transaction 1: {cur1.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2: {cur2.fetchall()}\\n\"\\\n",
    "                      f\"Changes visible to transaction 2.\"\\\n",
    "                     )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rollback Transactions\n",
    "\n",
    "The next example shows a similar transaction as above. The only difference is that instead of making the changes persistent, we decide to [`ABORT`](https://www.postgresql.org/docs/current/sql-abort.html) the transaction by calling `rollback()` on the connection. It is equivalent to running the following transaction directly from the database shell:\n",
    "```SQL\n",
    "BEGIN;\n",
    "UPDATE accounts SET balance = balance - 100 WHERE id=3;\n",
    "UPDATE accounts SET balance = balance + 100 WHERE id=1;\n",
    "ABORT;\n",
    "```\n",
    "All changes performed by the aborted must not become durable in the database. Note that if we `close()` an open connection, `rollback()` will be performed implicitly. In other words: if `autocommit` is set to `False`, calling `close()` is equivalent to aborting both the transaction and closing the connection."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Account balances observed by each connection before COMMIT:\n",
      "Transaction 1: [(1, 2200.0), (3, 270.0)]\n",
      "Transaction 2: [(1, 2100.0), (3, 370.0)]\n",
      "Changes not yet visible to transaction 2.\n",
      "--Transaction 1 aborted--\n",
      "Account balances observed by each connection after ABORT:\n",
      "Transaction 1: [(1, 2100.0), (3, 370.0)]\n",
      "Transaction 2: [(1, 2100.0), (3, 370.0)]\n",
      "Changes of transaction 1 undone.\n"
     ]
    }
   ],
   "source": [
    "with psycopg.connect(dsn) as conn1:\n",
    "    with psycopg.connect(dsn) as conn2:     \n",
    "        with conn1.cursor() as cur1:\n",
    "             with conn2.cursor() as cur2:\n",
    "                # Update balance of account 3, implicitly begins a transaction\n",
    "                cur1.execute(\"\"\"UPDATE accounts SET balance = balance - 100 WHERE id=3;\"\"\") \n",
    "                # Update balance of account 1, implicitly begins a transaction\n",
    "                cur1.execute(\"\"\"UPDATE accounts SET balance = balance + 100 WHERE id=1;\"\"\")\n",
    "                # Compare states visible to both transactions\n",
    "                q_acc = \"\"\"SELECT * FROM accounts WHERE id=1 OR id=3;\"\"\"\n",
    "                cur1.execute(q_acc)\n",
    "                cur2.execute(q_acc)\n",
    "                print(f\"Account balances observed by each connection before COMMIT:\\n\"\\\n",
    "                      f\"Transaction 1: {cur1.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2: {cur2.fetchall()}\\n\"\\\n",
    "                      f\"Changes not yet visible to transaction 2.\"\\\n",
    "                     )\n",
    "                # Abort transaction 1\n",
    "                conn1.rollback()\n",
    "                print(\"--Transaction 1 aborted--\")\n",
    "                # Compare states visible to both transactions\n",
    "                cur1.execute(q_acc)\n",
    "                cur2.execute(q_acc)\n",
    "                print(f\"Account balances observed by each connection after ABORT:\\n\"\\\n",
    "                      f\"Transaction 1: {cur1.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2: {cur2.fetchall()}\\n\"\\\n",
    "                      f\"Changes of transaction 1 undone.\"\\\n",
    "                     )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Isolation Levels\n",
    "\n",
    "Furthermore, we can set the isolation level ([`SET TRANSACTION`](https://www.postgresql.org/docs/current/sql-set-transaction.html)) per session by calling `set_isolation_level()` or `set_session()` with the corresponding parameters.\n",
    "\n",
    "The following example showcases the impact of isolation levels. While transaction 1 withdraws money from a bank account, transaction 2 sets its isolation level to `REPEATABLE READ` and reads the entry for this bank account. After transaction 1 commited, transaction 2 again accesses the same bank account. However, since its isolation level is set to `REPEATABLE READ` it still sees the unchanged data.\n",
    "\n",
    "The scenario is equivalent to running the following transactions in parallel from two database shells.\n",
    "\n",
    "Transaction 1:\n",
    "```SQL\n",
    "BEGIN;\n",
    "SELECT * FROM accounts WHERE id=2;\n",
    "UPDATE accounts SET balance = balance - 50 WHERE id=2;\n",
    "COMMIT;\n",
    "```\n",
    "\n",
    "Transaction 2:\n",
    "```SQL\n",
    "BEGIN;\n",
    "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;\n",
    "SELECT * FROM accounts WHERE id=2;\n",
    "-- in the meantime Transaction 1 updates account 2 and commits.\n",
    "SELECT * FROM accounts WHERE id=2;\n",
    "COMMIT;\n",
    "```\n",
    "\n",
    "Note that in SQL, isolation levels are set *within* a transaction block while `psycopg` requires us to set the isolation level *before* we start a new transaction. The isolation level supported by PostgreSQL and `psycopg` can be found [here](https://www.postgresql.org/docs/current/transaction-iso.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Account balance observed by each transaction:\n",
      "Transaction 1: [(2, 520.0)]\n",
      "Transaction 2: [(2, 520.0)]\n",
      "Both transactions see the same balance.\n",
      "--Update perfomed, Transaction 1 commited--\n",
      "Account balance observed by each transaction:\n",
      "Transaction 1: [(2, 470.0)]\n",
      "Transaction 2: [(2, 520.0)]\n",
      "Transaction 2 still sees the state from the beginning of the transaction.\n"
     ]
    }
   ],
   "source": [
    "with psycopg.connect(dsn) as conn1:\n",
    "    #  the default isolation level in PostgreSQL is READ COMMITTED\n",
    "    with psycopg.connect(dsn) as conn2:\n",
    "        conn2.isolation_level=psycopg.IsolationLevel.REPEATABLE_READ\n",
    "        with conn1.cursor() as cur1:\n",
    "            with conn2.cursor() as cur2:\n",
    "                # Compare states visible to both transactions\n",
    "                q_acc = \"\"\"SELECT * FROM accounts WHERE id=2;\"\"\"\n",
    "                cur1.execute(q_acc)\n",
    "                cur2.execute(q_acc)\n",
    "                print(f\"Account balance observed by each transaction:\\n\"\\\n",
    "                      f\"Transaction 1: {cur1.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2: {cur2.fetchall()}\\n\"\\\n",
    "                      f\"Both transactions see the same balance.\")\n",
    "                \n",
    "                # Withdraw money from account 2 and commit\n",
    "                cur1.execute(\"\"\"UPDATE accounts SET balance = balance - 50 WHERE id=2;\"\"\")\n",
    "                conn1.commit()\n",
    "                print(\"--Update perfomed, Transaction 1 commited--\")\n",
    "\n",
    "                # Compare states visible to both transactions\n",
    "                q_acc = \"\"\"SELECT * FROM accounts WHERE id=2;\"\"\"\n",
    "                cur1.execute(q_acc)\n",
    "                cur2.execute(q_acc)\n",
    "                print(f\"Account balance observed by each transaction:\\n\"\\\n",
    "                      f\"Transaction 1: {cur1.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2: {cur2.fetchall()}\\n\"\\\n",
    "                      f\"Transaction 2 still sees the state from the beginning of the transaction.\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.10"
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}