Example of Flask application with SQLAlchemy-IRIS - Part 1

Index

Part 1

• Introducing Flask: a quick review of the Flask Docs, where you will find all the information you need for this tutorial;
• Connecting to InterSystems IRIS: a detailed step-by-step of how to use SQLAlchemy to connect to an IRIS instance;

Part 2

• A discussion about this kind of implementation: why we should use it and situations where it is applicable.
• The OpenExchange Application: if you are not interested in the development of this example, you can go directly to this section and learn how to use it as a template for your flask-iris application.
• Bonus: some errors I have faced while developing and how I solved them.

Introducing Flask

Flask is a web framework written in Python and designed to make your life easier when creating a web application. It uses Werkzeug to deal with Web Server Gateway Interface (WSGI) - Python’s standard specification for web server communication. It also utilizes Jinja for handling HTML templates, and the Click CLI toolkit to create and manage command-line interface code.

Installation

The first step is optional. However, it is a good idea to set up a virtual environment before starting projects. You can do that on the terminal by changing the directory to the folder you want to use as the root and typing the command mentioned below.

> python -m venv .venv

If you choose to use a virtual environment, activate it with the following command (on Windows) before installing the prerequisites:

> .venv\Scripts\activate

Finally, install the packages with pip (Python install package).

> pip install Flask
> pip install Flask-SQLAlchemy
> pip install sqlalchemy-iris

Quickstart

We will follow the structure presented in this Tutorial, adapting the connections to the IRIS. It will minimize your problems when improving your application since you will be able to consult the Flask Documentation without conflicts. However, since Flask is very flexible, it does not require you to follow any layouts, and you can consider the following steps as mere suggestions.

The approach presented here is to use the applications as packages and define an app factory to cover large projects. Yet, you could still fully build the Flask app with only five lines of code, as shown below.

from flask import Flask

app = Flask(__name__)


@app.route('/')
def hello():
    return 'Hello, world!'

A sample application

This section is a walkthrough of the steps you need to take to create the application flask-iris. You can follow this tutorial if you have never used a web framework before. However, if you want to completely understand the theory of this kind of workflow, scan my previous article about Examples of working with IRIS from Django, in particular the image in the beginning. You will be pleased to notice that different web frameworks have very similar logic. Yet, in Flask we can define the URLs and the views together as groups, called blueprints.

The requirements are Flask, Flask-SQLAlchemy, and sqlalchemy-iris. Intuitively, the bridge between the web application and your IRIS instance is built!
 

The package

First of all, create or select the folder you want to use as root. Let's call it flaskr-iris. We need Python to understand this folder as a package, so we must create an __init__.py file. That is where we will put our app factory.

Step 1 - Creating the app factory

The application factory is nothing more than a function. Inside it, we have to define a Flask() instance (the app), set its configurations, connect it to the database, and register the blueprints if we choose to use them. We will explore more details about the blueprints later, so right now you do not need to worry about them.

The function should be called create_app(). Start by importing Flask and creating an instance with the __name__ argument (check the documentation for further details). Then map the config property with a secret key and the database URI. Use the format “dialect://username:password@host:port/NAMESPACE”, following the recommendations from SQLAlchemy. In this case, the dialect should be the ‘iris’, created by CaretDev. The username and password are the ones you use to connect to the InterSystems IRIS instance pointed by the host and port, and the namespace speaks for itself.

# __init__.py
from flask import Flask

def create_app():
    # create and configure the app
    app - Flask(__name__, instance_relative_config=True)
    
    app.config.from_mapping(
        SECRET_KEY = "dev", # override it with a random generation key when deploying
        SQLALCHEMY_DATABSE_URI = "iris://_SYSTEM:sys@loclhost:1972/SAMPLE"
    )

The upcoming lines of our factory will require the database, so let’s reserve this file for a minute while we talk about data and models.
 

Step 2 - Managing data

We will use SQLAlchemy to import and manage data since we can use Flask-SQLAlchemy and SQLAlchemy-IRIS to solve the connection problems. First, create the file database.py where we will import the Flask version of SQLAlchemy and instantiate it. This step does not require a separate file. However, it can come in handy later for developing more methods to deal with the database.

# database.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

Next, we must define the models that will tell Python how to interpret each record in a table. Create the file models.py, import the SQLAlchemy instance, and the features you need to treat data. Following the example given in the official Flask documentation, we can create a blog that will require one model for the users and another for the posts. This implementation covers a substantial number of cases and gives you a good head start.

We are going to use Object Relational Mapping (ORM) as suggested in SQLAlchemy 2.0, defining mapped columns and a One-to-many relationship. You can check how to model other types of relationships in SQLAlchemy’s guide.

# models.py
from .database import db
from sqlalchemy.orm import Mapped, mapped_column
from typing import List, Optional
import datetime

class User(db.Model):
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    username: Mapped[str] = mapped_column(db.String(1000), unique=True, nullable=False)
    password: Mapped[str] = mapped_column(db.String(1000), nullable=False)
    posts: Mapped[List["Post"]] =  db.relationship(back_populates="user")
    
class Post(db.Model):
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    author_id: Mapped[int] = mapped_column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    user: Mapped["User"] = db.relationship(back_populates="posts")
    created: Mapped[datetime.datetime] = mapped_column(db.DateTime, nullable=False, server_default=db.text('CURRENT_TIMESTAMP'))
    title: Mapped[str] = mapped_column(db.String(1000), nullable=False)
    body: Mapped[Optional[str]] = mapped_column(db.String(1000000))

The general format you can use for defining the columns on the latest SQLAlchemy release is:

columnName: Mapped[type] = mapped_column(db.Type, arguments)

As for the relationships, on the “many” side, you will also need to define the foreign key.

Obs.: the class names will be converted from “CamelCase” to “snake_case” when creating the SQL tables.

Finally, we can return to the factory on __init__.py and make the connections. Import the database and the models, assign this app to the db instance, and create all the tables inside the create_app() method.

# __init__.py
from flask import Flask
from sqlalchemy.exc import DatabaseError

def create_app():
    # create and configure the app
    app = Flask(__name__)
    
    app.config.from_mapping(
        SECRET_KEY="dev", # override with random when deploy
        SQLALCHEMY_DATABASE_URI = "iris://_SYSTEM:sys@localhost:1972/SAMPLE"
    )
    
    # flask initializes Alchemy with this app
    from .database import db
    from .models import User, Post
    
    db.init_app(app)
    
    try:
        with app.app_context():
            db.create_all()
    except DatabaseError as err:
        if 'already exists' in err._sql_message():
            print("Databases already exist.")
        else:
            print(err) 

Next, we will move on to creating the views.

Step 3 - The views

Step 3.1 - The blueprints: authentication

A Blueprint object was designed to organize your views in groups and register them in the application. We will create one Blueprint for the views that deal with authentication and another one for posting, editing, and deleting, so this part of the tutorial will cover CRUD and session management. 
Starting with the authentication blueprint, create the file auth.py and import the utilities from Flask and Werkzeug, the database, and the model User. Then, instantiate the Blueprint object, specifying its name and URL prefix.

# auth.py
from flask import (
    Blueprint, request, g, redirect, url_for, flash, render_template, session
)
from werkzeug.security import generate_password_hash, check_password_hash
from .database import db
from .models import User
import functools

bp = Blueprint('auth', __name__, url_prefix='/auth')

Next, we can use decorators to define routes for registering, logging in, and logging out with every method needed. In our example, we will employ GET and POST and start with registration.

The decorator is accessed as a method from our Blueprint and has one argument for the URL and another one for the methods it accepts. Right on the following line, create the function that tells the app what to do when it receives a request with the corresponding parameters. In the upcoming example, we will have access to the function by GETting or POSTing the URL http://host:port/auth/register.
 

# auth.py
@bp.route('/register', methods=('GET', 'POST'))
def register():
    if request.method=='POST':
        pass

If the app gets a POST referring to ‘/register’, it means that a user wants to create an account. Accessing the chosen username and password with the request content object is handled by Flask. Next, create an instance of the User model, storing its properties as the received values. You can also use the werkzeug.security methods to protect sensitive content. Then, utilize the session property of our database, provided by SQLAlchemy, to add the new user to the table and commit. At this point, the information has already been sent to the corresponding table in the IRIS instance referred to in the last step. Finally, redirect the user to the login page, which we will create in the next step. Do not forget to treat such errors as receiving empty inputs, index integrity, and connection to the database. You can use flash() to display details about the problem to the user. We can also utilize the same register() function to render the template for the register page if GET is used on the request.

# auth.py
@bp.route('/register', methods=('GET', 'POST'))
def register():
    if request.method=='POST':
        username = request.form['username']
        password = request.form['password']
        error = None
        if not username:
            error = "Username is required."
        elif not password:
            error = "Password is required."
        if error is None:
            try:
                user = User(
                    username=username,
                    password=generate_password_hash(password)
                )
                
                db.session.add(user)
                db.session.commit()
            except Exception as err: # TODO: change this to sqlite3's db.IntegrityError equivalent
                error = str(err)
            else:
                return redirect(url_for("auth.login"))
        
        flash(error)
    
    return render_template('auth/register.html')

Next, we repeat the same logic for the login page. When treating a POST, we first receive a username and password and then check if it exists in our IRIS table. This time, we will not use the session property (although we could have). Instead, we will apply the one_or_404() method. We should do it because the username must be unique, as we defined in our model. Thus, if the query does not return precisely one row as a result of our request, we can treat it as not found. Inside this function, chain SQL commands to find the required result, using the models as tables and their properties as columns. Finally, clear Flask’s session, add the user if the login was effective, and redirect them to the home page. If it is a GET, send the user to the login page.

#auth.py
@bp.route('/login', methods=('GET', 'POST'))
def login():
    if request.method=="POST":
        username = request.form['username']
        password = request.form['password']
        error = None
        try:
            user = db.one_or_404(db.select(User).where(User.username==username))
            
            if not check_password_hash(user.password, password):
                error = "Incorrect password"
        except Exception as err: # TODO also check this error
            error = f"User {username} not found."
        if error is None:
            session.clear()
            session['user_id'] = user.id 
            return redirect(url_for('index'))
        
        flash(error)
        
    
    return render_template('auth/login.html')
    

To log out, it is just enough to clear the session and redirect to the home page. 

# auth.py
@bp.route('/logout')
def logout():
    session.clear()
    return redirect(url_for('index'))

On the following functions, we are going to treat the user logging in on different requests, not necessarily authentication-related. The ‘g’ object, from Flask, is unique per each request, meaning you can use it to set the current user from the database when dealing with possibly forbidden accesses.

Fortunately, the Blueprint has the property before_app_request. It can be used as a decorator followed by a function to determine what to do when receiving a request before dealing with it. To load the logged user, we should perform the logging-in again, but this time, we will get information from the session other than the request.

# auth.py
@bp.before_app_request
def load_logged_in_user():
    user_id = session.get('user_id')
    
    if user_id is None:
        g.user = None
    else:
        g.user = db.one_or_404(
            db.select(User).where(User.id==user_id)
        )

Finally, for our last authentication function, we can make something that will deal with views that should be forbidden if the user has not logged in. If you do not want to display a simple 403 code, redirect the user to the login page.

# auth.py
def login_required(view):
    @functools.wraps(view)
    def wrapped_view(**kwargs):
        if session.get('user_id') is None:
            return redirect(url_for('auth.login'))
        
        return view(**kwargs)
    
    return wrapped_view

Step 3.2 - The blueprints: blog (CRUD)

Once again, start by creating the file (let's call it blog.py) and importing everything we will need. It may seem like a lot of work, but each of these imports will start making sense as we go. Also, create another Blueprint instance.

# blog.py
from flask import (
    Blueprint, flash, g, redirect, render_template, request, url_for, session
)
from werkzeug.exceptions import abort
from .auth import login_required
from .database import db
from .models import Post, User
from sqlalchemy import desc

bp = Blueprint('blog', __name__)

First, we should start by creating the default URL, pointing to the home page. Here, we will show the posts stored in IRIS, so we can use the session property of the database (which is an SQLAlchemy() instance from Flask-SQLAlchemy). There are many ways to execute a query with the SQLAlchemy module. I chose to use the function scalars() with the SQL statement as an argument, chained with the all() function. You may have seen similar behavior in other languages and plugins as fetchall(). Your next step will be to render the template for the home page, passing the variable where we stored all the posts as an argument so that we can access them in HTML files. Check out our READ (cRud) below:

# blog.py
@bp.route('/')
def index():
    posts = db.session.scalars(
        db.select(Post).order_by(desc(Post.created))
    ).all() 
    
    return render_template('blog/index.html', posts=posts)

Next, the function for creating (Crud) a new post will be very similar to registering a user. We get the user’s input from the request, set its values to the properties of the model Post, use the database’s session object to add (equivalent to the insert SQL statement), and commit. Here, we will access the current user from the request with the g object. Also, we will use a second decorator, referencing the function login_required(), to redirect the user to the login page if they have not logged in yet.

# blog.py
@bp.route('/create', methods=('GET', 'POST'))
@login_required
def create():
    if request.method == 'POST':
        title = request.form['title']
        body = request.form['body']
        error = None
        if not title:
            error = 'Title is required.'
        if error is not None:
            flash(error)
        else:
            post = Post(
                title=title,
                body=body,
                author_id=g.user.id
            )
            db.session.add(post)
            db.session.commit()
            
            return redirect(url_for('blog.index'))
        
    return render_template('blog/create.html')

To perform an update (crUd), we will use a very similar approach. However, instead of creating a new instance of Post, we will return it from a query, since it already exists in the database. We can use the URL to receive an ID for the Post. Once again, the login is required here.

# blog.py
@bp.route('/<int:id>/update', methods=('GET', 'POST'))
@login_required
def update(id):
    post = get_post(id)
    
    if request.method == 'POST':
        title = request.form['title']
        body = request.form['body']
        error = None
        if not title:
            error = 'Title is required.'
        if error is not None:
            flash(error)
        else:
            post.title = title
            post.body = body
            
            db.session.commit()
            
            return redirect(url_for('blog.index'))
    
    return render_template('blog/update.html', post=post)

Since for deleting we will also need to access the Post by the ID, we can create a function get_post(id) and use it on both views. Similarly to the other queries we have made so far, we can chain SQL commands to write it. Since our model for User has a relationship, we can select Post and join with User. In this way, we will be able to access the user’s information by the result object as well. We can also use abort() to send 404 (not found) and 403 (forbidden) responses.

# blog.py
def get_post(id, check_author=True):
    post = db.one_or_404(
        db.select(Post).join(User).where(Post.id == id)
    )
    
    if post is None:
        abort(404, f"Post id {id} doesn't exist.")
        
    if check_author and post.author_id != session['user_id']:
        abort(403)
        
    return post

Last but not least, you can perform the delete with nothing more than a delete function from the database session.

# blog.py
@bp.route('/<int:id>/delete', methods=('POST',))
@login_required
def delete(id):
    post = get_post(id)
    
    db.session.delete(post)
    db.session.commit()
    
    return redirect(url_for('blog.index'))

Step 3.3 - Register your blueprints

Finally, we can finish our app factory by registering the blueprints and returning the app, as shown below.

# __init__.py
from flask import Flask
from sqlalchemy.exc import DatabaseError

def create_app():
    # create and configure the app
    app = Flask(__name__)
    
    app.config.from_mapping(
        SECRET_KEY="dev", # override with random when deploy
        SQLALCHEMY_DATABASE_URI = "iris://_SYSTEM:sys@localhost:1972/SAMPLE"
    )
    
    # flask initializes Alchemy with this app
    from .database import db
    from .models import User, Post
    db.init_app(app)
    try:
        with app.app_context():
            db.create_all()
    except DatabaseError as err:
        if 'already exists' in err._sql_message():
            print("Databases already exist.")
        else:
            print(err) 
            
    # register blueprints
    from . import auth
    app.register_blueprint(auth.bp) 
    
    from . import blog
    app.register_blueprint(blog.bp)
    app.add_url_rule('/', endpoint='index')
    
    return app

Step 4 - The Templates

You must have noticed that every view finishing with a return redirect(), sends the user to another view or returns render_template(), which is self-explanatory. Those functions receive HTML templates as arguments, along with any other objects you might want to use inside them. It means that at this point of the tutorial, you will learn how to access your IRIS database from an HTML file, which will allow you to manage data with CSS and JavaScript, and evolve to any other web development tools. 

The default path for your Flask application built into a package to look for templates is in the app folder/templates. Inside it, it is customary to add a base.html file. The pattern for those files is the Jinja’s. That means you can use {% … %} to add such statements as blocks, “ifs”, and “fors”, always ending them (with {% endblock %}, for instance). You can also use {{ … }} for such expressions as accessing objects you passed in the render template function. Finally, you can use {# … #} for comments. 

Aside from the arguments passed in the views, you can access such context as the g object, which is unique per request, and get_flashed_messages() to display arguments passed in flash(). Since in the last steps, we have already seen how to make queries and pass the result into the request context, the following example for a base file will show you how to access data from IRIS using the g object. On top of that, it will also demonstrate how to display the flashed messages.
 

<!-- templates/base.html -->
<!DOCTYPE html>
<title>{% block title %}{% endblock %} - Flaskr</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
<nav>
    <h1>Flaskr</h1>
    <ul>
        {% if g.user %}
          <li><span>{{ g.user['username'] }}</span>
          <li><a href="{{ url_for('auth.logout') }}">Log Out</a>
        {% else %}
          <li><a href="{{ url_for('auth.register') }}">Register</a>
          <li><a href="{{ url_for('auth.login') }}">Log In</a>
        {% endif %}
    </ul>
</nav>
<section class="content">
    <header>
        {% block header %}{% endblock %}
    </header>
    {% for message in get_flashed_messages() %}
      <div class="flash">{{ message }}</div>
    {% endfor %}
    {% block content %}{% endblock %}
</section>

With the base being defined by now, you can finally create other pages by extending this file and personalizing the blocks. Explore the following example for the Create Posts page.

<!-- templates/blog/create.html -->
{% extends 'base.html' %}

{% block header %}
    <h1>{% block title %}New Post{% endblock %}</h1>
{% endblock %}

{% block content %}
    <form method="post">
        <label for="title">Title</label>
        <input name="title" id="title" value="{{ request.form['title'] }}" required>
        <label for="body">Body</label>
        <textarea name="body" id="body">{{ request.form['body'] }}</textarea>
        <input type="submit" value="Save">
    </form>
{% endblock %}

The name parameter in the input will be the key to the request.form dictionary, accessible on the views. The authentication pages login and register will have quite similar syntax since both only need one form. However, you can check them on my GitHub repository if you wish.

Now, let’s look at the home page, called index.html. 

<!-- templates/blog/index.html -->
{% extends 'base.html' %}

{% block header %}
 <h1>{% block title %}Posts{% endblock %}</h1>
 {% if g.user %}
    <a class="action" href="{{ url_for('blog.create') }}">New</a>
 {% endif %}
{% endblock %}

{% block content %}
    {% for post in posts %}
        <article class="post">
            <header>
                <div>
                    <h1>{{ post['title'] }}</h1>
                    <div class="about">by {{ post.user.username }} on {{ post['created'].strftime('%Y-%m-%d') }}</div>
                </div>
                {% if g.user['id'] == post['author_id'] %}
                    <a class="action" href="{{ url_for('blog.update', id=post['id'] )}}">Edit</a>
                {% endif%}
            </header>
            <p class="body">{{ post['body'] }}</p>
        </article>
        {% if not loop.last %}
            <hr>
        {% endif %}
    {% endfor %}
{% endblock %}

Here we have some <a/> elements worth taking a closer peek at. The link for creating a new post solely appears when g.user confirms the true, meaning that a user has logged in. There is also a “for” statement, iterating through the posts object returned by a query on the view for the empty URL “\” and passed to this file as a render_template() argument. For each post, we can access the username related to each user with a simple syntax (post.user.username), thanks to the db.relationship added in the User model. An alternative way to access data is to use the posts as dictionaries and display their properties with index syntax (such as post[‘title’]). Pay particular attention to how the link for updating a post passes its ID as an argument and only appears if its user matches the current logged one.

Finally, we have the update page, concurrently referring to the delete view. It is analogous to the Create Posts page, but it has a curious logic to define the default values from the database. I will leave an illustration below in case you decide to explore it.

<!-- templates/blog/udpate.html -->
{% extends 'base.html' %}

{% block header %}
    <h1>{% block title %}Edit "{{ post['title'] }}"{% endblock %}</h1>
{% endblock %}

{% block content %}
    <form method="post">
        <label for="title">Title</label>
        <input name="title" id="title" value="{{ request.form['title'] or post['title'] }}" required>
        <label for="body">Body</label>
        <textarea name="body" id="body">{{ request.form['body'] or post['body'] }}</textarea>
        <input type="submit" value="Save">
    </form>
    <hr>
    <form action="{{ url_for('blog.delete', id=post['id']) }}" method="post">
        <input type="submit" class="danger" value="Delete" onclick="return confirm('Are you sure?');">
    </form>
{% endblock %}

Part 5 - Adding static files

At this point, we have an interactive interface that receives information all the way from the user to the database and back, and we have IRIS objects accessible in Python with the help of SQLAlchemy and models made with class syntax. In the last step, we also managed to transform those objects into the content of HTML elements, meaning you can easily personalize and interact with them with CSS and JavaScript.

Taking a look back at the HTML base, check out the line with a link referencing a stylesheet, right in the beginning: 

<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">

The first argument for the url_for() function inside the href expression is the folder inside the root where you can find the statics. The second one, of course, is the file name. From now on, the design is up to you! Create the style.css file and edit as you please.

All done!

With all the files we have created, the root folder should look like the image below:

Now, go to the root directory on the terminal and type

flask --app flaskr-iris run --debug

Then follow the link http://127.0.0.1:5000 to see how everything is working.  

Conclusion

In this article, we have seen how Flask makes it possible and easy to bring information from an IRIS instance anywhere to a web application, transforming it into a fully personalizable CRUD, along with the option of connecting to other databases and systems. Now, you can easily create, display, update, and delete information from an IRIS database, via an interface that not only allows you to customize its appearance but also lets you treat data with any Python tool without hustle.

You can contact me to share any doubts or ideas you have encountered after reading! In the upcoming article, I will discuss more of the theory of this implementation and present the OpenExchange application developed here. As a bonus, I will reveal some errors I ran into and the solutions I managed to find in the last section.