Flask and SQLAlchemy

Creating a Many to Many Table

In my day job of working in IT for education, we have a difficult time tracking our asset inventory. With thousands of devices in the hands of staff and students, tracking them with spreadsheets and Google Forms becomes very tedious and almost a full time responsibility. We can't justify the cost of investing in an enterprise asset tracking suite of tools so I have been thinking of writing our own for some time. I have been working on it for the last few weeks and have been tackling some difficult (to me) problems. I am using the Flask framework to build a web app to track our technology assets and how they are assigned to users.

Due to my nature, I have viewed this project as a way to build coding skills along with building something useful. The web application has three major components: managing and tracking users, managing and tracking assets, and tracking notes/tickets related to the assets. The first big system I tackled was the user management module. In addition to registering and handling the login/logout flow for the app, I have to implement a system for giving users of the application privileges within the app. Users should be able to see what assets have been assigned to them and any history or notes associated with those assets, Agents should be able to add, delete, and edit users and assets, and admins should be able to do everything plus make systemic changes to the way the application is organized.

To assign and track users and their role in the system, an association table was needed to to keep track of which roles are assigned to which users. The code below shows how this was implemented within the Flask app. It should be noted that SQLAlchemy is used to manage the database connection and Flask-Login is managing user authentication.

The Data Models

In the user table, the role is related to one listed in the roles table. The role here, places the user id in a row with the role id by using the secondary="user_role_assignments" parameter in the relationship.

user.py

import datetime
import sqlalchemy as sa
import sqlalchemy.orm as orm

from flask_login import UserMixin
from asset_db.data.model_base import SqlAlchemyBase


class User(UserMixin, SqlAlchemyBase):
    # set table name
    __tablename__ = 'users'

    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    date_created = sa.Column(sa.DateTime, default=datetime.datetime.utcnow)
    staff_number = sa.Column(sa.Integer)
    f_name = sa.Column(sa.String, index=True)
    l_name = sa.Column(sa.String, index=True)
    status = sa.Column(sa.String)
    hashed_pw = sa.Column(sa.String)

    # get all the role related to the user
    role = orm.relationship("UserRole", lazy='subquery', secondary="user_role_assignments", back_populates="users")

    @property
    def full_name(self):
        return f"{self.f_name} {self.l_name}"

    def __repr__(self):
        return f'<User: {self.f_name} {self.l_name}>'

In the roles table, the user is related to one listed in the user table. The user here, places the role id in a row with the user id by using the secondary="user_role_assignments" parameter in the relationship.

user_role.py

import datetime
import sqlalchemy as sa
import sqlalchemy.orm as orm

from asset_db.data.model_base import SqlAlchemyBase


class UserRole(SqlAlchemyBase):
    # set table name
    __tablename__ = 'user_roles'

    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    date_created = sa.Column(sa.DateTime, default=datetime.datetime.utcnow)
    role = sa.Column(sa.String, nullable=False)

    users = orm.relation("User", secondary="user_role_assignments", back_populates="role")

    def __repr__(self):
        return f'<Role: {self.role}>'

To bring the two together, the association table keeps rows that have a user id and a role id together so users can see what role they have and you can see what users are assigned to a specific role.

user_role_assignment.py

import datetime
import sqlalchemy as sa
import sqlalchemy.orm as orm

from asset_db.data.model_base import SqlAlchemyBase
from asset_db.data.user import User
from asset_db.data.user_role import UserRole


class UserRoleAssignment(SqlAlchemyBase):
    # set table name
    __tablename__ = 'user_role_assignments'

    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    date_created = sa.Column(sa.DateTime, default=datetime.datetime.utcnow)
    user_id = sa.Column(sa.Integer, sa.ForeignKey("users.id"))
    role_id = sa.Column(sa.Integer, sa.ForeignKey("user_roles.id"))

    user = orm.relationship(User, backref=orm.backref("role_assignment", cascade="all, delete-orphan"))
    user_role = orm.relationship(UserRole, backref=orm.backref("role_assignment", cascade="all, delete-orphan"))

The View Functions

Finally, here are the Flask routes that are used to update a user account using a form to get information from the user. There are a lot of things going on here that reflect what I have learned about implementing an association table and all I can say is this works in this specific implementation with this specific configuration. This code is the result of reading many stack overflow questions, many blog posts, and lots and lots of documentation. It seems to be working but is certainly not the most efficient or best example of how to make this work. The code snippet below also assumes there is data in the database to update.

A few things, a full list of roles is passed to the update form on a GET request so it can be used to populate the <options> tags in a <select> input field in the form. For a POST request, first we check if an user has an existing role and if not, we assign the role first. This bit of code updates the account information for the user that is logged in so using the current_user provided by Flask-Login is a handy way to check. But, once the new role is written, the current_user is not reloaded so we have to grab an updated user object with the new role assigned or we will write a blank role on top of the newly assigned one, hence the use of user rather than current_user to update most of the information.

The need to check for the existence of a role is the result of how the database is set up. To get this to work in a development environment, test data is loaded into the database using a csv upload and the test data doesn't assign a role to the user (because I haven't figured out how to do that yet). As a result, the vast majority of the users don't have a role assigned to them. That is also why I think I need a separate database session here because it is a new row, I have to use session.add(user) here and then use session.merge(user) later because it is updating a row.

To update the relationship between user and role, I ran into a number of obstacles, too. The first was how to get the correct role from the form in a usable object to add to the association table. This is where passing a list of roles to the form was handy. The selected role will pass the correct role id when the form is submitted, but the role id is not what needs to be added to the association table (as many SQLAlchemy errors showed). What needs to be added to the association table is the UserRole object, so I used a helper function to get that object by looking it up in the database.

Then, I thought using the user.role.append(updated_role) would be all I needed but this will add a new row to the association table each time the form is submitted. This makes sense because what I am doing is using a many to many relationship so many users can be assigned to each role and many roles can be assigned to each user in this structure. But practically, this is not how I want the app to behave - each user can only have one role at a time. Ultimately there is probably a better, simpler way to implement this but - this is working for me now. To prevent users for having multiple roles in the association table, then, I used the user.role.clear() method to wipe out the old role before assigning the new. I don't know if that is the right way to do it, but it appears to be doing what it is supposed to be doing.

user_views.py

from flask import render_template, Blueprint, request, redirect, url_for, flash, abort
from flask_login import current_user, login_user, logout_user, login_required

from asset_db.services import user_service
from asset_db.data import db_session

users = Blueprint('users', __name__, template_folder='templates')

...

# ##########  ACCOUNT UPDATE  ##########

# view to update account information
@users.route('/account/update', methods=['GET'])
@login_required
def account_update_get():

    # get a list of available roles to assign a user
    roles = user_service.get_all_roles()

    return render_template('user/account-update.html', roles=roles)


# view to write changes to the database
@users.route('/account/update', methods=['POST'])
@login_required
def account_update_post():

    r = request

    # check if role exists and add if it doesn't
    if not current_user.role:
        user = user_service.find_user_by_id(current_user.id)
        new_role = user_service.find_role_by_id(r.form.get('role'))
        user.role.append(new_role)
        session = db_session.create_session()
        session.add(user)
        session.commit()
        session.close()

    user = user_service.find_user_by_id(current_user.id)

    user.staff_number = r.form.get('staff_number')
    user.f_name = r.form.get('f_name')
    user.l_name = r.form.get('l_name')
    user.email = r.form.get('email').lower().strip()
    user.phone1 = r.form.get('phone1')
    user.phone2 = r.form.get('phone2')
    user.organization = r.form.get('organization')
    user.site = r.form.get('site')
    user.department = r.form.get('department')
    user.position = r.form.get('position')
    # remove existing role from association table
    user.role.clear()
    # add new role to association table
    updated_role = user_service.find_role_by_id(r.form.get('role'))
    user.role.append(updated_role)
    user.status = r.form.get('status')

    # update the user information in the db
    session = db_session.create_session()
    session.merge(user)
    session.commit()
    session.close()

    # Send the user back to the account profile page
    flash("You have successfully updated your profile!", 'success')
    resp = redirect(url_for("users.account_profile"))

    return resp

Here are the helper methods in the service module:

user_service.py

from typing import Optional, List

from asset_db.data import db_session
from asset_db.data.user import User
from asset_db.data.user_role import UserRole

# ##########   USERS   ##########

def get_all_roles() -> List[UserRole]:
    session = db_session.create_session()
    roles = session.query(UserRole).all()
    session.close()
    return roles


# ##########   ROLES   ##########

def find_role_by_id(role_id: int) -> Optional[UserRole]:
    session = db_session.create_session()
    role = session.query(UserRole).filter(UserRole.id == role_id).first()
    session.close()
    return role

So, after a few weeks of work and many, many blogs, tutorials, and videos my web application is taking shape and I am getting close to having the user management portion of the application functioning more and more like I had hoped.

Resources

Talk Python Training - Specifically the "Building Data-Driven Web Apps with Flask and SQLAlchemy" course
Corey Schaefer YouTube - Specifically his series on Flask tutorials
Flask Mega Tutorial by Miguel Grinberg
SQLAlchemy Relationships Documentation
Michael Cho Blog Post
Tutorials Point Many to Many Relationships