Python DB connector: script / tutorial to connect to a database and perform simple operations on a table using psycopg2

dbconnector.py

import psycopg2
import re


def connect():
    """
    Function to connect to the server
    NOTE: this, as it is, works only on my machine. You need to insert your own connection data
    :return: psycopg2 'connection' object to the university server
    :rtype: psycopg2 'connection' object
    """

    # fetch params for the server connection from a text file
    with open("../../server_uni_connection_data.txt", 'r') as connectionDataFile:
        connectionDataDict = {"host": None, "db": None, "user": None, "password": None}
        for line in connectionDataFile.readlines():
            lineData = line.split(':')
            connectionDataDict[lineData[0]] = lineData[1].strip()  # remove final '\n'

    # connect to the database
    return psycopg2.connect(host=connectionDataDict["host"],
                            database=connectionDataDict["database"],
                            user=connectionDataDict["user"],
                            password=connectionDataDict["password"])


def display_menu():
    """
    Functions that display a menu with the possible options
    :return: a list of tuple (option name, corresponding function name) and the number of options
    :rtype: list[ tuples(str, function name) ], int
    """
    options = [
        ("Insert new expense", insertExpense),
        ("Check expenses history", seeExpenses),
        ("Delete an expense", deleteExpense)
    ]

    print("Insert the number of the desired option\nAvailable options:")
    count = 0
    for option in options:
        print(f"{count}) {option[0]}", end="\n")
        count += 1

    print(f"{count}) Exit")
    return options, count


def register_choice(n_opts):
    """
    Function to register user's choice
    :param n_opts: number of options
    :type n_opts: int
    :return: the choice of the user
    :rtype: int
    """
    choice = -1
    while choice < 0 or choice > n_opts:
        choice = int(input("> "))

    return choice


def insert_date():
    """
    Function to insert a date from console and do the checks
    :return: the date typed by the user
    :rtype: string
    """
    day_per_month = {"01": 31, "02": 28, "03": 31, "04": 30, "05": 31, "06": 30, "07": 31, "08": 31, "09": 30,
                     "10": 31, "11": 30, "12": 31}

    # date format dd/mm/yyyy
    date = input("Insert date (dd/mm/yyyy): ")
    pattern = re.compile("^\d{2}/\d{2}/\d{4}$")
    while not pattern.match(date) or int(date[3:5]) > 12 or int(date[3:5]) < 1 or \
            int(date[0:2]) > day_per_month[date[3:5]] or int(date[0:2]) < 1:
        date = input("Wrong format! Insert date (dd/mm/yyyy): ")

    return date


def insertExpense(connection, cursor):
    """
    Function to insert new data in table expenses
    :param connection: instance of <class 'psycopg2.extensions.connection'>
    :param cursor:  instance of <class 'psycopg2.extensions.cursor'>
    """
    connection.isolation_level = 'READ COMMITTED'
    date = insert_date()
    description = input("Insert description: ")
    amount = float(input("Insert amount: "))
    cursor.execute("INSERT INTO Expenses(date, description, amount) VALUES(%s, %s, %s)", (date, description, amount))
    print("Insertion tuple: {:s}".format(cursor.statusmessage))
    if connection.notices:
        print(f"Notices: {connection.notices[-1]}")
        connection.notices[-1] = ""


def seeExpenses(connection, cursor):
    """
    Function to see all expenses
    :param connection: instance of <class 'psycopg2.extensions.connection'>
    :param cursor:  instance of <class 'psycopg2.extensions.cursor'>
    """
    connection.isolation_level = 'READ COMMITTED'
    cursor.execute("SELECT * FROM Expenses;")
    for record in cursor:
        print(record, end="\n")


def deleteExpense(connection, cursor):
    """
    Function to see all expenses
    :param connection: instance of <class 'psycopg2.extensions.connection'>
    :param cursor:  instance of <class 'psycopg2.extensions.cursor'>
    """
    connection.isolation_level = 'REPEATABLE READ'  # to avoid exit status 0 while a concurrent transaction deleted
    # that record. In that case the record might be actually deleted while having exit status 0 in this transaction

    seeExpenses(connection, cursor)
    id = input("Insert the id of the expense to delete: ")
    cur.execute("DELETE FROM Expenses WHERE id = %s;", id)
    print("DELETE: Exit status: {:s}\nAny notices: {:s}".format(cursor.statusmessage, connection.notices[-1]))
    connection.notices[-1] = ""


if __name__ == '__main__':
    con = connect()
    con.autocommit = True
    with con:
        with con.cursor() as cur:
            # create table "Expenses" if does not exist
            cur.execute("CREATE TABLE IF NOT EXISTS Expenses( \
                        id SERIAL PRIMARY KEY, \
                        date DATE NOT NULL, \
                        description VARCHAR, \
                        amount DECIMAL(10, 2) NOT NULL);")
            print("Creation table 'Expenses': {:s}".format(cur.statusmessage))
            if con.notices:
                print(f"Notices: {con.notices[-1]}")
                con.notices[-1] = ""  # to avoid repetition of the same notice in other operations

            # options_funcs_map: list[ tuples(str, func name) ]
            options_funcs_map, options_number = display_menu()
            choice = register_choice(options_number)

            while choice != options_number:  # if choice == option_number then exit
                # call the func which name is the 2nd value of the choice-th tuple of options_funcs_map
                options_funcs_map[choice][1](con, cur)
                print()
                options_funcs_map, options_number = display_menu()
                choice = register_choice(options_number)

    con.close()