This worksheet focuses on making web applications with Flask and SQLite.
The main references for these topics are:
You need the source code for Yellaro and Whinge to do your work this week. Download this zip file of all the source code and sample database files (as of 2021-04-17):
(The code is also on GitHub, but downloading the zip is the easiest way to get everything you need without downloading the entire course sample code repository.)
Get Yellaro and its database reset utility working
yellaro.py and check that you can view its main page and post messages using your web browser.yellaro.db in the zip file. If you didn't have a database file, it would be necessary to run dbreset.py first.yellaro.db and try to run/use Yellaro. Does it start up? Can it serve the main page?Get Whinge and its database reset utility working
whinge.py and check that you can view its main page, submit a new whinge, and change scores on items using the + and - buttons.dbreset.py for Whinge makes a blank database by default, but can add sample data upon request. The usage message it prints on the terminal describes how to do this.dbreset.py for Yellaro¶Modify the dbreset.py in Yellaro to behave similarly to the one from Whinge, where the default behavior is to clear or create the database and leave it empty, but it can also add a few rows of sample data if requested by a command line argument.
# MCS 275 Week 14 Problem 2
# J Vaccaro
# I modified the original yellaro dbreset function myself, in accordance with the syllabus.
"""Modified to add the new sample data."""
import sqlite3
# Add import sys
import sys
con = sqlite3.connect("yellaro.db")
con.execute("""
DROP TABLE IF EXISTS messages;
""")
con.execute("""
CREATE TABLE messages (
id INTEGER PRIMARY KEY,
sender TEXT,
content TEXT,
ts REAL
);
""")
# At this point, we have created a fresh, empty table.
# Add check for '--sampledata' as a command line arg. If it is one, then add some data.
if "--sampledata" in sys.argv:
# Update the user on what the program is doing
print("Adding sample data to table 'messages'.")
con.execute("""
INSERT INTO messages (sender,content,ts) VALUES
(?,?,?);
""",
("David","How's it going?",1618247723.27))
con.execute("""
INSERT INTO messages (sender,content,ts) VALUES
(?,?,?);
""",
("Zoe","Great, I just started working on Worksheet 13!!!",1618248002.79))
else:
# Print helpful messages to help with usage
print("Leaving table 'messages' empty. (Run this script with '--sampledata' as a command line argument to add some sample data.)")
con.commit()
con.close()
Modify Yellaro so that when displaying the contents of a message, any time the word Python appears (with any capitalization), it is shown in a different color than the rest of the message text. Do this by detecting the presence of the word in the data retrieved from the database, wrapping it in a <span class="py"> tag, and then styling that class of span tags in the CSS. (You shouldn't save the modified text to the database.)
If you want to go furhter and have the color change over time, here is some sample CSS that does that. The text in any span tag of class py will fade between two of the main colors in the Python language logo:
span.py {
color: #3B8BBE;
animation: colorcycle 3s infinite;
}
@keyframes colorcycle {
0% {
color: #3B8BBE;
}
50% {
color: #FFD43B;
}
100% {
color: #3B8BBE;
}
}# MCS 275 Week 14 Problem 2 -- Add these lines to yellaro.db
# J Vaccaro
# I wrote this code myself, in accordance with the syllabus
"""Added function python_format, modified function"""
def python_replace(content):
"""Adds span tags around any mention of python (any combo of upper/lowercase)"""
content_lower = content.lower()
start_tag = "<span class='py'>"
end_tag = "</span>"
# This will track our offset in indices between the original content and our modified version
ind = 0
# Iterates over potential starting indices for a python substring in content
# Warning!!! uses the forbidden construction
for i in range(len(content_lower)-5):
# Check whether the (lowercased) substring is python
if content_lower[i:i+6] == "python":
# Insert the tags around the python mention.
content = content[0:i+ind] + start_tag + content[i+ind:i+6+ind] + end_tag + content[i+6+ind:]
# Update the index offset with the lengths of the added tags
ind += len(start_tag) + len(end_tag)
# Return the content string, with tags added.
return content
# You must modify message_div, because you only want to change the display, not the data saved in the database.
def message_div(row):
"""Take a sqlite3.Row with columns "sender" and "content" and convert it
to a single HTML div that is compatible with the stylesheet."""
return """<div class="message"><span class="message-username">{}</span><span class="message-content">{}</span></div>\n""".format(
row["sender"],
python_replace(row["content"]) # Call the function with adds the span tags around 'python'
)
Add a route /new/ to Whinge which is similar to the current front page, but which sorts submissions from newest to oldest instead of sorting by score. Limit the number of submissions that are shown. It should have the same +/- buttons for changing the score of an item, and should have the submission form at the bottom.
Doing so will introduce some rough edges to the user interface, where any action you take on the new posts page (clicking + or -, submitting a new post) will take you back to /top/, because all parts of the application that perform actions are hard-coded to redirect back there when they finish. At first, that's fine, but you should then fix this problem as follows:
The URL that is the target of the +/- buttons should include a second query parameter in addition to postid, which will tell the application whether it should redirect back to /new/ or /top/ after updating the score. The top and new pages generate appropriate URLs when they render the list of submissions. The /plus and /minus routes should be modified to check this parameter and act accordingly.
The new post submission form at the bottom of each page should have a hidden input field that stores the page the user was on when they composed the submission (either /new/ or /top/). The route that handles form submission uses this to decide where to redirect the client after the form data are processed.
# MCS 275 Week 14 Problem 3
# J Vaccaro and D Dumas
# We modified whinge.py following syllabus rules.
"""Whinge with both /new/ and /top/ pages"""
from flask import Flask, url_for, request, redirect
import sqlite3
import datetime
import time
app = Flask(__name__)
HEADER="""<!DOCTYPE html>
<html>
<head>
<title>Whinge</title>
<link rel="stylesheet" href="/static/whinge.css">
</head>
<body>
<h1>Whinge</h1>
"""
FOOTER="""</div>
<div class="compose-form">
<form action="/post" method="post">
<div>
<label for="username">Username:</label>
<input type="text" id="username" name="username">
</div>
<div>
<label for="username">Whinge:</label>
<input type="text" id="whinge" name="whinge" size="80">
</div>
<input type="hidden" name="page" value="{}">
<input type="submit" value="Submit">
</form>
</div>
</body>
</html>
"""
def footer(current_page):
"""Insert current_page as the value of hidden form field page in the
FOOTER template above (so that the route which handles that form
submission will redirect back to current_page)."""
return FOOTER.format(current_page)
def get_db():
"""Open a connection to the database and return the connection object"""
con = sqlite3.connect("whinge.db")
con.row_factory = sqlite3.Row # return dict-like rows
return con
def post_div(row,current_page):
"""Take a sqlite3.Row from the posts table and format it as a div compatible
with the stylesheet."""
timestr = datetime.datetime.fromtimestamp(row["ts"]).strftime("%Y-%m-%d %H:%M")
return """
<div class="post">
<div class="post-main">
<span class="post-score">{0}</span><a class="post-plus-button" href="{4}">+</a><a class="post-minus-button" href="{5}">-</a><span class="post-content">{1}</span>
</div>
<div class="post-metadata">
Submitted by <span class="post-author">{2}</span> at <span class="post-timestamp">{3}</span>
</div>
</div>
""".format(
row["score"],
row["content"],
row["submitter"],
timestr,
url_for("score_plus_one",postid=row["postid"],page=current_page),
url_for("score_minus_one",postid=row["postid"],page=current_page)
)
@app.route("/")
def root_redirect():
"""Root resource redirects to the top posts display"""
return redirect(url_for("display_top"))
@app.route("/top/")
def display_top():
"""Show the top-ranked posts"""
con = get_db()
document = HEADER
document += """
<div class="navigation">
Showing <strong>top</strong> posts. [Show <a href='/new/'>new</a>]
</div>"""
for row in con.execute("SELECT * FROM posts ORDER BY score DESC LIMIT 10;"):
document += post_div(row,"/top/")
con.close()
document += footer("/top/")
# return the complete HTML document
return document
@app.route("/new/")
def display_new():
"""Show the top-ranked posts"""
con = get_db()
document = HEADER
document += """
<div class="navigation">
Showing <strong>new</strong> posts. [Show <a href='/top/'>top</a>]
</div>"""
for row in con.execute("SELECT * FROM posts ORDER BY ts DESC LIMIT 10;"):
document += post_div(row,"/new/")
con.close()
document += footer("/new/")
# return the complete HTML document
return document
@app.route("/post",methods=["GET","POST"])
def create_post():
"""Receive form data and add a row to the database"""
# Check for and reject empty username or whinge
if not request.values.get("username") or not request.values.get("whinge"):
print("Ignoring request to with empty username or whinge")
else:
# Form data ok; add to DB
con = get_db()
con.execute("INSERT INTO posts (submitter,content,ts) VALUES (?,?,?);",
(
request.values.get("username"), # form field username -> DB column submitter
request.values.get("whinge"), # form field whinge -> DB column content
time.time()
)
)
con.commit()
con.close()
# TODO: Handle possibility of failed INSERT
# Send them back to whatever page they came from when submitting the form
# (default="/" means that if we somehow received a form that doesn't have
# the page field set, we redirect back to "/")
return redirect(request.values.get("page",default="/"))
@app.route("/plus") # /plus?postid=15 means add 1 to score for postid 15
def score_plus_one():
"""Find a post by `postid` (given as query parameter) and
increase its score by one."""
# Retrieve the query parameter postid
postid = request.values.get("postid")
con = get_db()
con.execute("""
UPDATE posts SET score=score+1 WHERE postid=?;
""",
(postid,))
con.commit()
con.close()
return redirect(request.values.get("page"))
@app.route("/minus")
def score_minus_one():
"""Find a post by `postid` (given as query parameter) and
decrease its score by one."""
# TODO: Avoid duplicated code with score_plus_one; have
# both call a single add_to_score function.
postid = request.values.get("postid")
con = get_db()
con.execute("""
UPDATE posts SET score=score-1 WHERE postid=?;
""",
(postid,))
con.commit()
con.close()
return redirect(request.values.get("page"))
if __name__=="__main__":
app.run()
Choose Yellaro or Whinge for this. The concept can be applied to either one. In this problem you'll add features which treat messages and/or users differently if a message or submission contains certain words. (This is meant as a way to test your skills in implementing interesting program logic, not as an endorsement of content filtering or censorship.)
Make a global variable in the main Python program that contains a list of words that are hidden when they appear in posts. The words in this list should be replaced by a series of * characters when displaying the post, but the database should still hold the original text.
e.g. if the list is LEVEL1_BANNED_WORDS = ["C++","Java"], an attempt to post
Before learning Python, I spent a while experimenting with Java.then the application should store this but display it as
Before learning Python, I spent a while experimenting with ****.Add a second list of words that, if found in a post, result in the post not being added to the database at all. Optionally, display a message to the user indicating that something went wrong.
e.g. if the list is LEVEL2_BANNED_WORDS = ["blockchain","home equity loan"], then a user would be unable to post the following message:
I think blockchain technology is going to revolutionize grading of programming assignments! Want to hear how?Add a third list of words that, if found in a post, result in the user who submitted the post being added to a database table of banned users. The post does not appear, nor does any future post from that username. (Previous posts by this user can still be shown.)
e.g. if the list is LEVEL3_BANNED_WORDS = ["durian","pineapple pizza"], and if user ddumas attempts to post the message
I was so full from eating that entire durian fruit that I didn't have any room for dinner.then no further messages submitted by user ddumas will be accepted.
(To do this, you'll need to alter the database by adding a new table. I suggest updating dbreset.py so that it creates all the tables your application needs, even if you initially create the new table in the sqlite command line shell or with the Python REPL.)
Also make command line utilities that can be used to manually ban or unban a user specified as a command line argument, e.g.
python3 unban.py ddumas # unbanned; durian is not so bad
python3 ban.py ddumas # banned again; starting talking about the HYG star databaseNote: The right way to add support for user-level actions like this is to have a separate table listing all users, assigning each one a unique id number, and storing characteristics like username, ban status, etc.. The table of messages or posts would then refer to the user by id number, rather than storing the username. But for this problem, you can use a simpler implementation where the table for banned users has a single column of type TEXT and acts like a persistent list of strings.
# dbreset.py
# MCS 275 Spring 2021 David Dumas
# Chat app (Yellaro) database creation/reset
"""Reset the yellaro database; version supporting user bans"""
import sqlite3
import sys
con = sqlite3.connect("yellaro.db")
# delete messages table
con.execute("""
DROP TABLE IF EXISTS messages;
""")
# create messages table
con.execute("""
CREATE TABLE messages (
id INTEGER PRIMARY KEY,
sender TEXT,
content TEXT,
ts REAL
);
""")
# delete ban table
con.execute("""
DROP TABLE IF EXISTS banned_users;
""")
# create ban table
# (making the username a primary key forces it to be unique)
con.execute("""
CREATE TABLE banned_users (
username TEXT PRIMARY KEY
);
""")
if "--sampledata" in sys.argv[1:]:
con.execute("""
INSERT INTO messages (sender,content,ts) VALUES
(?,?,?);
""",
("David","How's it going?",1618247723.27))
con.execute("""
INSERT INTO messages (sender,content,ts) VALUES
(?,?,?);
""",
("Zoe","Great, I just started working on Worksheet 13!!!",1618248002.79))
con.execute("""
INSERT INTO banned_users VALUES
(?);
""",
("Benedict Arnold",))
con.commit()
con.close()
# yellaro.py
# MCS 275 Spring 2021 - David Dumas
"""Simple chat web app with post filtering and banning"""
from flask import Flask, url_for, request, redirect
import sqlite3
import time
LEVEL1_BANNED_WORDS = ["C++","Java"]
LEVEL2_BANNED_WORDS = ["blockchain","home equity loan"]
LEVEL3_BANNED_WORDS = ["durian","pineapple pizza"]
app = Flask(__name__)
# Flask automatically handles URLs beginning with /static
# by looking for files in the /static subdirectory of the
# applicaiton directory. We use this to deliver the CSS.
HEADER="""<!DOCTYPE html>
<html>
<head>
<title>Yellaro</title>
<link rel="stylesheet" href="/static/yellaro.css">
</head>
<body>
<h1>Yellaro</h1>
<div class="messagebox">
"""
# footer format string has placeholders for the default username {0}
# and the URL that should be used for posting a message {1}
FOOTER="""</div>
<div class="refresh-form">
<form action="/" method="get">
<input type="submit" value="Check for new messages">
<input type="hidden" name="lastuser" value="{0}">
</form>
</div>
<div class="message-compose-form">
<form action="{1}" method="post">
<div>
<input type="text" id="message" name="message" class="fullwidth">
</div>
<div>
<label for="username">Username:</label>
<input type="text" id="username" name="username" value="{0}">
</div>
<input type="submit" value="Send">
</form>
</div>
</body>
</html>
"""
def footer():
"""Front page footer (the end of the messagebox div and everything after it).
This contains the message composition form. It needs to be dynamically generated
because of the feature that auto-populates the username with the previously-used
one if it is provided as a query param `lastuser`, and because it uses `url_for`
to get the URL of the form submission target."""
return FOOTER.format(
request.values.get("lastuser",default=""),
url_for("post_message") # url_for takes a *function* name and
# determines which URL calls it. Here
# URL /post calls post_message.
)
def get_db():
"""Open a connection to the yellaro database
and return the connection object"""
con = sqlite3.connect("yellaro.db")
con.row_factory = sqlite3.Row # return dict-like rows
return con
def format_content(s):
"""Take message content `s` and format for display in the UI, obscuring
words from LEVEL1_BANNED_WORDS"""
# TODO: Use regular expressions to only replace *entire* words, e.g.
# so that banning "car" does not ban "carpet". But for this worksheet we
# do a simple substring replace
# TODO: Also handle case-insensitive match
for w in LEVEL1_BANNED_WORDS:
s = s.replace(w,"*"*len(w))
return s
def message_div(row):
"""Take a sqlite3.Row with columns "sender" and "content" and convert it
to a single HTML div that is compatible with the stylesheet."""
return """<div class="message"><span class="message-username">{}</span><span class="message-content">{}</span></div>\n""".format(
row["sender"],
format_content(row["content"])
)
@app.route("/")
def message_feed():
"""Return the HTML message feed and new message form."""
con = get_db()
feed = ""
# We only show 10 messages, so they need to be the 10 largest timestamps.
# Here we fetch them all, resulting in a list with the newest message first
rows = con.execute("SELECT sender,content FROM messages ORDER BY ts DESC LIMIT 10;").fetchall()
# But we want to display them with oldest first, so we reverse the list
for row in reversed(rows):
feed += message_div(row)
con.close()
# return the complete HTML document
return HEADER + feed + footer()
def ban_user(con,username):
"""Add username to the table of banned users"""
# We don't want many copies of a single user name in the table.
# One approach would be to test whether it is already there
# with a SELECT and then add it with INSERT if the SELECT returned
# nothing. That was the solution I expected students to come up
# with on this worksheet.
# Here we demonstrate another way to do it. We created the banned_users
# table with `username` as a PRIMARY KEY, so it won't allow two rows
# to have the same `username` value. The INSERT command can take an
# optional qualifier OR IGNORE that specifies the command should do
# nothing if the INSERT attempt would violate a constraint. This
# means a single SQL statement can handle the change.
con.execute("INSERT OR IGNORE INTO banned_users VALUES (?);",
(username,)
)
con.commit()
def failpage(reason,backurl):
"""Display a page for a failed mesage post operation, with a link
back to the main page"""
return """<!DOCTYPE html>
<html>
<head>
<title>Yellaro</title>
<link rel="stylesheet" href="/static/yellaro.css">
</head>
<body>
<h1>Yellaro</h1>
<div>
<strong>Posting error: {}</strong>
</div>
<div>
<a href="{}">Click here to return to the message feed.</a>
</div>
</body>
</html>
""".format(reason,backurl)
@app.route("/post",methods=["GET","POST"])
def post_message():
"""Receive form data and add a row to the database"""
# Whether called after HTTP GET or POST, the form fields
# are available with `flask.request.values.get(fieldname)`
# Since we used `from flask import request` earlier, we
# can shorten this to `request.values(fieldname)`
# Unpack form data into conveniently-named variables
username = request.values.get("username")
message = request.values.get("message")
# Returning main_redir anywhere will send user back to the main page
# with the username field in the form pre-filled
main_url = url_for("message_feed",lastuser=username)
main_redir = redirect(main_url)
if not message or not username:
print("Ignoring request to post message with empty content or username")
return main_redir
con = get_db()
# Check for problems with the message in decreasing order of severity
# User is already banned?
banned = con.execute("SELECT COUNT(*) FROM banned_users WHERE username=?;",
(username,)
).fetchone()[0] # fetchone returns a tuple like (1,) so [0] extracts the actual count
# Now banned is 1 (truthy) if they are banned, else 0 (falsy)
if banned:
con.close()
return failpage("""
User '{}' is permanently banned from posting.
""".format(username),main_url)
# Bannable content?
for w in LEVEL3_BANNED_WORDS:
if w in message:
ban_user(con,username)
con.close()
return failpage("""
Because you attempted to post a message containing
'{}', you have been permanently banned from posting.
""".format(w),main_url)
# Content that prevents a message from appearing?
for w in LEVEL2_BANNED_WORDS:
# TODO: Improve this to a proper word check, not just substring
if w in message:
con.close()
return failpage("""
'{}' is not allowed to appear in messages.
""".format(w),main_url)
# User not banned, content acceptable. Add the message.
con.execute("INSERT INTO messages (sender,content,ts) VALUES (?,?,?);",
(
username, # form field username -> DB column sender
message, # form field message -> DB column content
time.time()
)
)
con.commit()
con.close()
return main_redir
if __name__=="__main__":
app.run()
# unban.py
# MCS 275 Spring 2021 David Dumas
"""Un-ban a user from yellaro (command line tool)"""
import sqlite3
import sys
if len(sys.argv)<2:
print("Usage: {} USER_TO_UNBAN".format(sys.argv[0]))
exit(1)
username = sys.argv[1]
con = sqlite3.connect("yellaro.db")
con.execute("DELETE FROM banned_users WHERE username=?;",
(username,)
)
con.commit()
con.close()
print("unbanned '{}'".format(username))
# ban.py
# MCS 275 Spring 2021 David Dumas
"""Ban a user from yellaro (command line tool)"""
import sqlite3
import sys
if len(sys.argv)<2:
print("Usage: {} USER_TO_BAN".format(sys.argv[0]))
exit(1)
username = sys.argv[1]
con = sqlite3.connect("yellaro.db")
# See comments in function ban_user of yellaro.py for
# details on "OR IGNORE" and another way to add a row
# only if it isn't there already.
con.execute("INSERT OR IGNORE INTO banned_users VALUES (?);",
(username,)
)
con.commit()
con.close()
print("banned '{}'".format(username))
Steps to test all the new features:
dbreset.py --sampledata and then yellaro.pyJava is easier to read than Python. as any user (see censored version)Can I interest you in information about a home equity loan? as any user (receive error page)Anybody know a discreet way to contact the British admiralty? AFAF as user Benedict Arnold (receive error message because this user is already banned in the sample data created by dbreset.pyHey everyone as user David (succeeds)I left a bunch of sliced durian on the common room table in case anyone is hungry as user David (receive error message; David is now banned)Sorry, I meant *churros*. I left churros on the table. as user David (receive error message; David is banned from posting)unban.py DavidDavid again and it should work.