Building the data layer with PostgreSQL

PostgreSQL is an object-relational database system, and Flask can utilize it as a data storage platform if the activated virtual environment has the psycopg2-binary extension module. To install this extension module into the venv, run the following command:

pip install psycopg2-binary

Now, we can write an approach to establish a connection to the PostgreSQL database.

Setting up database connectivity

There are multiple ways to create a connection to a database, but this chapter will showcase a Pythonic way to extract that connection using a custom decorator. In the project’s /config directory, there is a connect_db decorator that uses psycopgy2.connect() to establish connectivity to the opcs database of our prototype. Here is the implementation of this custom decorator:

import psycopg2
import functools
from os import environ
def connect_db(func):
    @functools.wraps(func)
    def repo_function(*args, **kwargs):
        conn = psycopg2.connect(
            host=environ.get('DB_HOST'),
            database=environ.get('DB_NAME'),
            port=environ.get('DB_PORT'),
            user = environ.get('DB_USER'),
            password = environ.get('DB_PASS'))
        resp = func(conn, *args, **kwargs)
        conn.commit()
        conn.close()
        return resp
    return repo_function

The given decorator provides the connection instance, conn, to a repository function and commits all the changes to the database after a transaction’s successful execution. Also, it will close the database connection at the end of the process. All the database details, such as DB_HOST, DB_NAME, and DB_PORT, are stored as environment variables inside a .env file. To retrieve them using the environ dictionary of the os module, run the following command to install the required extension:

pip install python-dotenv

However, there are other ways to manage these custom and built-in configuration variables instead of storing them as .env variables. The next topic will expound on this, but first, let’s apply @connect_db to our repository layer.

Implementing the repository layer

The following insert_signup() transaction adds a new user signup record to the database. It gets the conn instance from the @connect_db decorator. Our application has no object-relational mapper yet and solely depends on the psycopg2 driver to perform the CRUD operation. The cursor instance created by conn executes the INSERT statement of the following transaction with form data provided by its view function:

from config.db import connect_db
from typing import Dict, Any, List
@connect_db
def insert_signup(conn, user:str, passw:str, utype:str, fname:str, lname:str, cid:str) -> bool:
    try:
        cur = conn.cursor()
        sql = 'INSERT INTO signup (username, password, user_type, firstname, lastname, cid) VALUES (%s, %s, %s, %s, %s, %s)'
        values = (user, passw, utype, fname, lname, cid)
        cur.execute(sql, values)
        cur.close()
        return True
    except Exception as e:
        cur.close()
        print(e)
    return False

cursor is an object derived from conn that uses a database session to perform insert, update, delete, and fetch operations. So, just like insert_signup(), the following transaction uses cursor again to execute the UPDATE statement:

@connect_db
def update_signup(conn, id:int, details:Dict[str, Any]) -> bool:
    try:
        cur = conn.cursor()
        params = ['{} = %s'.format(key) for key in details.keys()]
        values = tuple(details.values())
        sql = 'UPDATE signup SET {} where id = {}'.format(', '.join(params), id);
        cur.execute(sql, values)
        cur.close()
        return True
    except Exception as e:
        cur.close()
        print(e)
    return False

To complete the CRUD operations for the signup table, here is the DELETE transaction from our application:

@connect_db
def delete_signup(conn, id) -> bool:
    try:
        cur = conn.cursor()
        sql = 'DELETE FROM signup WHERE id = %s'
        values = (id, )
        cur.execute(sql, values)
        cur.close()
        return True
    except Exception as e:
        cur.close()
        print(e)
    return False

The use of an ORM to build the model layer will be part of Chapter 2’s discussions. For now, the views and services of our application rely on a repository layer that manages PostgreSQL data directly through the psycopg2 driver.

After creating the repository layer, many applications can build a service layer to provide loose coupling between the CRUD operations and the views.

Creating the service layer

The service layer of the application builds the business logic of the view functions and the repository. Instead of loading the view functions with transaction-related and business processes, we place all these implementations in the service layer by creating lists of all the counselor and patient IDs, validating where to persist the newly approved user, and creating a list of patients who excelled in the examinations. The following service function evaluates and records patients’ exam scores:

def record_patient_exam(formdata:Dict[str, Any]) -> bool:
    try:
        pct = round((formdata['score'] / formdata['total']) * 100, 2)
        status = None
        if (pct >= 70):
            status = 'passed'
        elif (pct < 70) and (pct >= 55):
            status = 'conditional'
        else:
            status = 'failed'
        insert_patient_score(pid=formdata['pid'], qid=formdata['qid'], score=formdata['score'], total=formdata['total'], status=status, percentage=pct)
        return True
    except Exception as e:
        print(e)
    return False

Instead of directly accessing insert_patient_score() to save patient exam scores, record_score() accesses the record_patient_exam() service to compute some formulas before invoking insert_patient_score() from the repository layer for record insertion. The service lessens some friction between the database transactions and the view layer. The following snippet is the view function that accesses the record_patient_exam() service for record exam record insertion:

@app.route('/exam/score', methods=['GET', 'POST'])
def record_score():
    if request.method == 'GET':
        pids = list_pid()
        qids = list_qid()
        return render_template( 'exam/add_patient_score_form.html', pids=pids, qids=qids), 200
    else:
        params = dict()
        params['pid'] = int(request.form['pid'])
        params['qid'] = int(request.form['qid'])
        params['score'] = float(request.form['score'])
        params['total'] = float(request.form['total'])
        result = record_patient_exam(params)
        … … … … … … …
        else:
            return redirect('/exam/task/error')

Aside from calling record_patient_exam(), it also utilizes the list_pid() and list_qid() services to retrieve the IDs. The use of services can help separate the abstraction and use cases from the route functions, which has a beneficial impact on the scope, clean coding, and runtime performance of the routes. Moreover, the project structure can also contribute to clear business flow, maintainability, flexibility, and adaptability.