Demo
Contributors: Alicia Wang, Conner Swenberg
SQLite3 Demo
Create a Table
Creating a table containing rows of users with an ID, name, age, and optional email.
CREATE TABLE user (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
age INT NOT NULL,
email CHAR(50)
);Creating a table containing rows of fruit with an ID, name, and whether or not it is healthy.
CREATE TABLE fruit (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
is_healthy INTEGER NOT NULL
);See Table Columns
To see that we created the columns that we intended!
$ PRAGMA table_info(fruit);See Tables
To verify that our tables have actually been added to our database!
.tables;Delete a Table
Delete the fruit table.
DROP TABLE fruit;Add Data Into a Table
Add a user named Alanna with age 20 and email [email protected].
INSERT INTO user (name, age, email)
VALUES ("Alanna", 20, "[email protected]"); Add two more users to our database, Conner and Alicia.
INSERT INTO user (name, age, email)
VALUES ("Conner", 21, "[email protected]");
INSERT INTO user (name, age)
VALUES ("Alicia", 21);Retrieve Data From a Table
Get data from only two columns from our user table, name and email.
SELECT name, email FROM user;Get all data from our user table.
SELECT * FROM user;Get data from two columns, name and age, that satisfies the condition that the user is 18 years old or older.
SELECT name, age FROM user WHERE age >= 18;Use AND or OR, SQL keywords, to chain conditionals.
SELECT id, name, age FROM user WHERE name = "Alicia" AND age > 20;Update Data in a Table
Replace the email for only the user with ID as 1, which happens to be Alanna.
UPDATE user SET email = "[email protected]" WHERE id = 1;Delete Data from a Table
Delete users who are 21 years old from our user table, which happens to be Conner and Alicia. You can verify this with SELECT * FROM user;.
DELETE FROM user WHERE age = 21;File Setup
import os
import json
import sqlite3
class DatabaseDriver(object):
"""
Database driver for the Task app.
Handles with reading and writing data with the database.
"""
def __init__(self):
pass
First we will begin by setting up a new file, db.py, to house our database implementation. Notice how we are importing the sqlite3 package in line 3.
Our implementation approach is to construct this DatabaseDriver object as our interface for manipulating a SQLite database. We will define many methods within this class to allow us to read and write to our tables and abstract the calling of these methods in app.py.
Now back in app.py we want to instantiate our database driver as a variable as follows:
import json
from flask import Flask, request
import db
DB = db.DatabaseDriver()
app = Flask(__name__)We first import db.py at the top and then instantiate an instance of our DB driver class like any other python class. We will interact with our database within app.py through calling methods of this object. Note: if you are continuing on the same files from last week's demo, you can remove the initialization of our task_id_counter and tasks dictionary as these are being replaced the use of a real database.
Database methods
Database Initialization
To begin, let us complete our init function:
def __init__(self):
self.conn = sqlite3.connect("todo.db")
self.create_task_table()Connect to the Database: Our first line that sets self.conn represents our connection to the database. You can imagine calling sqlite.connect() as analogous to opening an Excel spreadsheet, loading in the contents of our tables for reading and manipulation. We are connecting to a database called "todo.db", thus implying that our database can be stored locally as a single file, which is a convenient way for us to store data for our purposes. By default, SQLite ensures that the database connection can only be used in the same thread where it was created, which helps prevent potential issues when working in multi-threaded environments. Since our course doesn't involve multithreading, we don't need to modify this behavior.
Initialize our Tables: Our next and final step for initializing our database is to create whatever tables we want to work with. For right now, we are calling a method create_task_table() that we have actually not defined yes, so let's do that:
def create_task_table(self):
try:
self.conn.execute("""
CREATE TABLE task (
ID INTEGER PRIMARY KEY AUTOINCREMENT,
DESCRIPTION TEXT NOT NULL,
DONE BOOLEAN NOT NULL
);
""")
except Exception as e:
print(e)Firstly, we wrap our function in a try-except so that if the table already exists, we can still proceed normally. We will be using self.conn.execute() often throughout our implementation and the function simply allows us to execute a SQL command as provided by a python string (we are using triple quotes, """ to write multi-line strings). As the function name indicates, we will be creating a table to store our tasks that contains the same fields as our previous task app implementation. As noted in lecture, each column definition involves providing a field name and column type. We can also provide the PRIMARY KEY flag on our ID field which tells SQLite that this is must be unique across the table and needs to be incremented as we create new rows in the table. The NOT NULL flag indicates that a column cannot contain Null values, i.e. it must contain some real information.
Get All Tasks
Thinking back to our application's routes from app.py the first query we want to implement is a method to get all tasks. We can do this as follows within our DB driver class:
def get_all_tasks(self):
cursor = self.conn.execute("SELECT * FROM task;")
tasks = []
for row in cursor:
tasks.append({"id": row[0], "description": row[1], "done": row[2]})
return tasksSelect Data with a Cursor: Using the same Excel table analogy, we can think of a query like dragging our cursor across many rows and columns. In our executed query, we wish to grab all columns (star * syntax) within the task table. We deliberately name this variable cursor to think of it like our cursor as we visually drag across the table. Executing this query will return a SQLite.Cursor object, but we can play with it like it is a 2D array (a list of lists).
Iterate over Rows: Now to parse this information, we can loop over each row within this 2D array-like object and construct a new python object by mapping values in the row-list to their appropriate field name. Column ordering is preserved, so row[0] is the first column ID, row[1] the second column DESCRIPTION, and row[2] the third column DONE. We can construct a new list to store these new objects, which will be our final return.
Add a Task
To add new tasks to our database, we will functionally be inserting new rows into our task table. We can define a method that takes a description and done argument and adds a new task with these attributes to our database:
def insert_task_table(self, description, done):
cursor = self.conn.cursor()
cursor.execute("INSERT INTO task (DESCRIPTION, DONE) VALUES (?, ?);",
(description, done))
self.conn.commit()
return cursor.lastrowidCreate a Cursor: We can also make SQL operations by first instantiating a cursor with self.conn.cursor()and then executing a SQL command with this cursor. In our previous operations, we were actually taking a shortcut by writing self.conn.execute. which would instantiate a cursor, execute the SQL command, and return us this cursor all in one function call.
Execute the Insert Command: To execute our insert command, we need to define the table we are inserting into, the columns we are providing values for, and our intended values. Notice that we do not define ID as a column we are providing a value for because one will be generated for us. In SQLite, we can mark placeholder values in our python string with ?. We can then provide the actual values inside a tuple as a second argument for the execute method.
Save and Return: Just like making a change to an Excel spreadsheet needs to be saved, so does the insert operation on our database. We save any changing to the database with the commit method. Upon committing our change, we will return the id of the last row available to the cursor (the task we just added). This will make more sense when we walk through how we call this method in app.py.
Retrieve a Task
To retrieve a specific task from our database, we will define a method that takes an id argument write a SQL query to execute:
def get_task_by_id(self, id):
cursor = self.conn.execute("SELECT * FROM task WHERE ID = ?", (id,))
for row in cursor:
return {"id": row[0], "description": row[1], "done": row[2]}
return NoneExecute the Retrieval Query: Our SQL query will select all columns (*) from our task table for rows that have their ID column equal to our provided ID. Notice how we use the same ? placeholder syntax and a tuple containing our values to substitute. Note that (id,) defines a tuple of length one with the seemingly extraneous comma; python will not evaluate this to a tuple if the comma is missing.
Return: Because we know that the ID field is a primary key in the task table, we know that this query will either return one row of data if our task exists, or none at all otherwise. Because of this we want to return on the first row loop of our cursor. If the cursor object contains no data, our for loop will act like it is looping over an empty list (therefore no loops) and we will return None.
Update a Task
To update a specific task in our database, we will utilize SQL's UPDATE keyword. We can define a method that takes an id, description, and done argument to identify our task of interest and update the fields appropriately:
def update_task_by_id(self, id, description, done):
self.conn.execute("""
UPDATE task
SET description = ?, done = ?
WHERE id = ?;
""", (description, done, id))
self.conn.commit()Execute the Update Command: To translate our SQL to plain english: we wish to UPDATE rows within the task table by setting the description and done columns to our values WHERE the task's id equals our id of interest. Just like inserting a new task into the table, our update command makes a change to the database and thus needs to be committed for the changes to persist.
Delete a Task
To delete a specific task in our database, we will utilize SQL's DELETE keyword. We can define a method that takes an id argument to identify our task of interest and remove it:
def delete_task_by_id(self, id):
self.conn.execute("""
DELETE FROM task
WHERE id = ?;
""", (id,))
self.conn.commit()Execute the Delete Command: Just like previously, do not forget to commit your change to the database!
Updates to Route Logic
Now we can transition back to app.py to call these database methods from within our routes. But first, we will abstract our response structure to clean up our code from last time.
Generalized Response Functions
app = Flask(__name__)
def success_response(body, code=200):
return json.dumps(body), code
def failure_response(message, code=404):
return json.dumps({'error': message}), codeAs you noticed in our last demo and assignment, we use json.dumps(...) a lot. Our responses follow an adaptable general structure, so we can turn generating these response payloads into a function.
For returning a successful response, we most often use the code 200 so we will set this as our default. Likewise, for failure responses, we most often use the code 404 because we cannot find a particular resource.
Get All Tasks
@app.route("/tasks/")
def get_tasks():
return success_response(DB.get_all_tasks())Retrieving all of our tasks is now as simple as calling our previously written method on our database driver. By abstracting interactions with the database to another file, we will dramatically clean up our application code. Everything else about how we define a route, construct a result dictionary, and return a JSON format of this data along with a response code remains the same from what we previously learned.
Post a Task
@app.route("/tasks/", methods=["POST"])
def create_task():
body = json.loads(request.data)
description = body["description"]
task_id = DB.insert_task_table(description, False)
task = DB.get_task_by_id(task_id)
if task is None:
return failure_response("Something went wrong while creating task!", 500)
return success_response(task, 201)Just like before, we need to parse our incoming request's body for the description data. We will can add a new task to the database with our insert_task_table method and will call get_task_by_id after to make sure our task was properly added. If our task was properly created, i.e. is not None, then we can return it and now use a response code of 201 to indicate a new resource was created.
Retrieve a Task
@app.route("/tasks/<int:task_id>/")
def get_task(task_id):
task = DB.get_task_by_id(task_id)
if task is None:
return failure_response("Task not found!")
return success_response(task)Just like before, we design this route to accept an id within the route path. All that has changed is that we call our database method for retrieving the task, which will return None if there does not exist a task with the provided id.
Update a Task
@app.route("/tasks/<int:task_id>/", methods=["POST"])
def update_task(task_id):
body = json.loads(request.data)
description = body["description"]
done = bool(body["done"])
DB.update_task_by_id(task_id, description, done)
task = DB.get_task_by_id(task_id)
if task is None:
return failure_response("Task not found!")
return success_response(task)Upon parsing the request's body, we will call our custom update method to make the execution. We want to return the newly updated data, so to make sure things worked properly, we retrieve our task from the database with a new query.
Delete a Task
@app.route("/tasks/<int:task_id>/", methods=["DELETE"])
def delete_task(task_id):
task = DB.get_task_by_id(task_id)
if task is None:
return failure_response("Task not found!")
DB.delete_task_by_id(task_id)
return success_response(task)If our task does exist, we will execute the deletion, otherwise we return a failed response.
Last updated
Was this helpful?