SQLAlchemy is a Object Relational Mapper library that provides an abstraction to working with databases. Instead of executing SQL queries you can manipulate database data by using Object Oriented programming code.

Task 1

Just as in lab 1 follow the link below to open repl.it. Fork the project then sign in.

Open Lab 2

Models are python classes which eventually become database tables. Objects created from models reflect a table row record.

When you open the workspace and view the contents of models.py. You should find a user class which is specified according to SQLAlchemy's documentation.

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)

SQLAlchemy's Model Documentation gives details on other data-types, constraints and setting up relationships with other models.

Model Methods

Models can also contain methods. We will need methods on our classes to perform some common operations on its instances. We shall define 3 methods:

  1. __init__(): This is the constructor of the class, used to make class instances
  2. set_password(): Because this is the model which handles user accounts we define this method which would hash user passwords
  3. __repr__(): This method returns a String representation of an instance of the model used for printing objects to the console.

Task 2

Add the following methods to User

def __init__(self, username, email, password):
  self.username = username
  self.email = email
  self.set_password(password)
  
def set_password(self, password):
    """Create hashed password."""
    self.password = generate_password_hash(password, method='sha256')

def __repr__(self):
    return f'<User {self.username} - {self.email}>'

Now our user model can now exhibit state and behaviour via its properties and methods.

Now that our models are created we can start saving data in our application. When you open wsgi.py you will see a script for a cli application using click. We can add various functionalities to our command line application by defining click commands. There is an init command already available for us.

wsgi.py

import click, sys
from models import db, User
from app import app


@app.cli.command("init", help="Creates and initializes the database")
def initialize():
  db.drop_all()
  db.init_app(app)
  db.create_all()
  bob = User('bob', 'bob@mail.com', 'bobpass')
  db.session.add(bob)
  db.session.commit()
  print(bob)
  print('database initialized')

Task 3

Test the command by running "flask init" in the replit shell (not console).

This should initialize the database and create the user.

We can query objects from our database using the query property available on any Model. Model.query can be used as follows:


Where <Model> is any class defined in models.py eg: User.query.get(1)

Task 4.1

Lets add a command to retrieve a user by their username and print it.

@app.cli.command("get-user", help="Retrieves a User")
@click.argument('username', default='bob')
def get_user(username):
  bob = User.query.filter_by(username=username).first()
  if not bob:
    print(f'{username} not found!')
    return
  print(bob)

Task 4.2

We can also create a command to get all user objects using query.all()

@app.cli.command('get-users')
def get_users():
  # gets all objects of a model
  users = User.query.all()
  print(users)

Now we can see all of the users of the application.

To make changes to an object simply reassign a new value to the desired property then save the object to the database.

Task 5

Add a command to update the email of a user.

@app.cli.command("change-email")
@click.argument('username', default='bob')
@click.argument('email', default='bob@mail.com')
def change_email(username, email):
  bob = User.query.filter_by(username=username).first()
  if not bob:
      print(f'{username} not found!')
      return
  bob.email = email
  db.session.add(bob)
  db.session.commit()
  print(bob)

Now we can call the command to update bob's email.

As our usernames and email have the unique constraint, we cannot reuse usernames or emails for new users. It is important that our apps anticipate such edgecases of otherwise valid input and fail gracefully instead of crashing.

These failure cases are called exceptions and we perform exception handling so that our application can fail gracefully and provide a useful message to the user.

Task 6

Implement a create-user command but handle any errors thrown by the database due to unique constraint violations.

@app.cli.command('create-user')
@click.argument('username', default='rick')
@click.argument('email', default='rick@mail.com')
@click.argument('password', default='rickpass')
def create_user(username, email, password):
  newuser = User(username, email, password)
  try:
    db.session.add(newuser)
    db.session.commit()
  except IntegrityError as e:
    #let's the database undo any previous steps of a transaction
    db.session.rollback()
    # print(e.orig) #optionally print the error raised by the database
    print("Username or email already taken!") #give the user a useful message
  else:
    print(newuser) # print the newly created user

Now we can safely add new users to the application. (note rick is the default for create-user)

Deletion is also easily done by simply using the db.session.delete() method then calling db.session.commit() to save the changes to the database.

Task 7

Create a command that will delete a user using db.session.delete()

@app.cli.command('delete-user')
@click.argument('username', default='bob')
def delete_user(username):
  bob = User.query.filter_by(username=username).first()
  if not bob:
      print(f'{username} not found!')
      return
  db.session.delete(bob)
  db.session.commit()
  print(f'{username} deleted')

As usual we always search if the username supplied actually exists. Now we can delete users from the application.

Now we want to make our second model which saves the To-Dos in our To-Do application. Because To-Do's must belong to a logged in user we need to set up a 1 to many relationship between a User and a To-Do.

We want to build a todo application, try to create a model that follows the table structure below. (Constructor and repr method implied). A model diagram is given below to show the relationship between todo and user. An association arrow is used to link the Todo.user_id foreign key to User.id


Most of the model would be very similar to user however, todo as a foreign key, this implemented as shown below

Task 8.1

Implement the Todo model according to the spec above. Note toggle() is a method that switches the boolean state of the done field..

class Todo(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) #set userid as a foreign key to user.id 
  text = db.Column(db.String(255), nullable=False)
  done = db.Column(db.Boolean, default=False)

  def toggle(self):
    self.done = not self.done
    db.session.add(self)
    db.session.commit()

  def __init__(self, text):
      self.text = text

  def __repr__(self):
    
    return f'<Todo: {self.id} | {self.user.username} | {self.text} | { "done" if self.done else "not done" }>'

We use db.ForeignKey to link this table to User. This way we establish a 1 to Many relationship from User to Todo.

Relationship Fields

Before we test, we are going to add relationship fields to our models. This is a powerful abstraction that makes all related objects available from an object of a different model. i.e. User objects compose their corresponding todo objects.

Task 8.2

Update User in models.py to add a todos field

class User(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  username = db.Column(db.String(80), unique=True, nullable=False)
  email = db.Column(db.String(120), unique=True, nullable=False)
  password = db.Column(db.String(120), nullable=False)
  #creates a relationship field to get the user's todos
  todos = db.relationship('Todo', backref='user', lazy=True, cascade="all, delete-orphan")

  def __init__(self, username, email, password):
    self.username = username
    self.email = email
    self.set_password(password)
    
  def set_password(self, password):
      """Create hashed password."""
      self.password = generate_password_hash(password, method='sha256')

  def __repr__(self):
      return f'<User {self.id} {self.username} - {self.email}>'

We use db.relationship to create our relationship field, the backref parameter creates a user field in todo so that every todo object can the user that created it in todo.user.

Now we can update our Model Diagram using composition arrows to indicate relationship fields. Note the foreign key association arrows have been removed as they are redundant because relationship fields require them.

Task 8.3

Update wsgi.py to import our Todo model.

wsgi.py

import click, sys
from models import db, User, Todo
from app import app

Then the init command to create and add a todo to our default user.

wsgi.py

@app.cli.command("init", help="Creates and initializes the database")
def initialize():
  db.drop_all()
  db.init_app(app)
  db.create_all()
  bob = User('bob', 'bob@mail.com', 'bobpass')
  bob.todos.append(Todo('wash car'))
  db.session.add(bob)
  db.session.commit()
  print(bob)
  print('database intialized')

Notice how we can use bob's todos relationship field to add to his list of todos. Secondly, we save our changes to bob which will create the appropriate todo record.

Task 8.4

Now we just need another command to see the todos of the user add the following:

wsgi.py

@app.cli.command('get-todos')
@click.argument('username', default='bob')
def get_user_todos(username):
  bob = User.query.filter_by(username=username).first()
  if not bob:
      print(f'{username} not found!')
      return
  print(bob.todos)

Finally we can now reinitialize the database, add our dummy data and view the todos of our user.

While the init file adds a default todo for us we can also make a command to create them individually..

Task 9.1

Implement a add todo command

wsgi.py

@app.cli.command('add-todo')
@click.argument('username', default='bob')
@click.argument('text', default='wash car')
def add_task(username, text):
  bob = User.query.filter_by(username=username).first()
  if not bob:
      print(f'{username} not found!')
      return
  new_todo = Todo(text)
  bob.todos.append(new_todo)
  db.session.add(bob)
  db.session.commit()

Test the command

Task 9.2

Next we implement a command that lets users toggle the done state of their todo.

@click.argument('todo_id', default=1)
@click.argument('username', default='bob')
@app.cli.command('toggle-todo')
def toggle_todo_command(todo_id, username):
  user = User.query.filter_by(username=username).first()
  if not user:
    print(f'{username} not found!')
    return

  todo = Todo.query.filter_by(id=todo_id, user_id=user.id).first()
  if not todo:
    print(f'{username} has no todo id {todo_id}')

  todo.toggle()
  print(f'{todo.text} is {"done" if todo.done else "not done"}!')

Note: we must ensure that the todo being updated actually belongs the the user hence user_id is checked in filter_by()

Finally we test

Often we may need to model many to many relationships in your applications. For example, let's add a category model such that todos can belong to many categories and a category can contain many todos.

Task 10.1

Add the following models to models.py

class TodoCategory(db.Model):
  __tablename__ ='todo_category'
  id = db.Column(db.Integer, primary_key=True)
  todo_id = db.Column(db.Integer, db.ForeignKey('todo.id'), nullable=False)
  category_id = db.Column(db.Integer, db.ForeignKey('category.id'), nullable=False)
  last_modified = db.Column(db.DateTime, default=func.now(), onupdate=func.now())

  def __repr__(self):
    return f'<TodoCategory last modified {self.last_modified.strftime("%Y/%m/%d, %H:%M:%S")}>'
  

class Category(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
  text = db.Column(db.String(255), nullable=False)
  user = db.relationship('User', backref=db.backref('categories', lazy='joined'))
  todos = db.relationship('Todo', secondary='todo_category', backref=db.backref('categories', lazy=True))

  def __init__(self, user_id, text):
    self.user_id = user_id
    self.text = text
  
  def __repr__(self):
    return f'<Category user:{self.user.username} - {self.text}>'

We can now update the Todo model so we can see their categories.

Task 10.2

Update the Todo class's __repr__ method as follows in models.py

def __repr__(self):
    category_names = ', '.join([category.text for category in self.categories])
    return f'<Todo: {self.id} | {self.user.username} | {self.text} | { "done" if self.done else "not done" } | categories [{category_names}]>'

Task 10.3

Then add some commands to categorize todos.

@click.argument('username', default='bob')
@click.argument('todo_id', default=6)
@click.argument('category', default='chores')
@app.cli.command('add-category', help="Adds a category to a todo")
def add_todo_category_command(username, todo_id, category):
  user = User.query.filter_by(username=username).first()
  if not user:
    print(f'{username} not found!')
    return

  res = user.add_todo_category(todo_id, category)
  if not res:
    print(f'{username} has no todo id {todo_id}')
    return

  print('Category added!')

Now we can categorize our todos

Thus concludes your introduction to flask-sqlalchemy. The usage of this library is at the very core of this course.

You can view a completed version of this lab at the following link

Open Lab 2 Completed

Resources