2.4 HACKS
Using Programs with Data is focused on SQL and database actions. Part A focuses on SQLAlchemy and an OOP programming style,
- Imports and Flask Objects
- Model definition
- Initial Data
- Check for credentials
- Create
- Read
- Update
- Delete
- Sqlite
"""
These imports define the key objects
"""
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
"""
These object and definitions are used throughout the Jupyter Notebook.
"""
# Setup of key Flask object (app)
app = Flask(__name__)
# Setup SQLAlchemy object and properties for the database (db)
database = 'sqlite:///sqlite.db' # path and filename of database
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = database
app.config['SECRET_KEY'] = 'SECRET_KEY'
db = SQLAlchemy()
# This belongs in place where it runs once per project
db.init_app(app)
""" database dependencies to support sqlite examples """
import datetime
from datetime import datetime
import json
from sqlalchemy.exc import IntegrityError
from werkzeug.security import generate_password_hash, check_password_hash
''' Tutorial: https://www.sqlalchemy.org/library.html#tutorials, try to get into a Python shell and follow along '''
# Define the User class to manage actions in the 'users' table
# -- Object Relational Mapping (ORM) is the key concept of SQLAlchemy
# -- a.) db.Model is like an inner layer of the onion in ORM
# -- b.) User represents data we want to store, something that is built on db.Model
# -- c.) SQLAlchemy ORM is layer on top of SQLAlchemy Core, then SQLAlchemy engine, SQL
""" database dependencies to support sqliteDB examples """
# Define the Score class to manage actions in the 'score' table
class Review(db.Model):
__tablename__ = 'favrecipes'
# Define the User schema with "vars" from object
id = db.Column(db.Integer, primary_key=True)
_recipelink = db.Column(db.String(255), unique=True, nullable=False)
_name = db.Column(db.String(255), unique=False, nullable=False)
_recipename = db.Column(db.String(255), unique=False, nullable=False)
_rating = db.Column(db.Integer, unique=False, nullable=False)
_comments = db.Column(db.String(255), unique=False, nullable=True)
# constructor of a User object, initializes the instance variables within object (self)
def __init__(self, recipelink, recipename, name, rating, comments):
self._recipelink = recipelink # variables with self prefix become part of the object,
self._recipename = recipename
self._name = name
self._rating = rating
self._comments = comments
#recipelink getter
@property
def recipelink(self):
return self._recipelink
#recipelink setter
@recipelink.setter
def recipelink(self, recipelink):
self._recipelink = recipelink
def is_recipelink(self, recipelink):
return self._recipelink == recipelink
#name setter
@property
def name(self):
return self._name
#name getter
@name.setter
def name(self, name):
self._name = name
#recipename getter
@property
def recipename(self):
return self._recipelink
#recipename setter
@recipename.setter
def recipename(self, recipename):
self._recipename = recipename
# rating getter
@property
def rating(self):
return self._rating
#rating setter
@rating.setter
def rating(self, rating):
self._rating = rating
#comments getter
@property
def comments(self):
return self._comments
#comments setter
@comments.setter
def comments(self, comments):
self._comments = comments
@property
def __str__(self):
return json.dumps(self.read())
def create(self):
try:
# creates a person object from Score(db.Model) class, passes initializers
db.session.add(self) # add prepares to persist person object to Users table
db.session.commit() # SqlAlchemy "unit of work pattern" requires a manual commit
return self
except IntegrityError:
db.session.remove()
return None
# CRUD read converts self to dictionary
# returns dictionary
def read(self):
return {
"id": self.id,
"recipelink": self.recipelink,
"recipename": self.recipename,
"name": self.name,
"rating": self.rating,
"comments": self.comments,
}
# CRUD update: updates user name, password, phone
# returns self
def update(self, name="", rating="", comments=""):
"""only updates values with length"""
if len(name) > 0:
self.name = name
if len(rating) > 0:
self.rating = rating
if len(comments) > 0:
self.comments = comments
db.session.commit()
return self
# CRUD delete: remove self
# None
def delete(self):
db.session.delete(self)
db.session.commit()
return None
"""Database Creation and Testing """
# Builds working data for testing
def initReview():
with app.app_context():
"""Create database and tables"""
db.create_all()
"""Tester data for table"""
u1 = Review(recipelink='https://sallysbakingaddiction.com/lemon-meringue-pie/', recipename= "Lemon Meringue Pie", name='Shruthi', rating='8', comments='Really good recipe, would definitely recommend')
u2 = Review(recipelink='https://www.foodnetwork.com/recipes/food-network-kitchen/apple-pie-recipe-2011423', recipename= "Apple Pie", name='Claire', rating='8', comments='Crust was very smooth')
u3 = Review(recipelink='https://www.loveandlemons.com/brownies-recipe/', recipename= "Brownies", name='Grace', rating='9', comments='So delicious')
u4 = Review(recipelink='https://addapinch.com/the-best-chocolate-cake-recipe-ever/', recipename= "Chocolate Cake", name='Jasmine', rating='9', comments='Thought the filling was perfect')
u5 = Review(recipelink='https://joyfoodsunshine.com/the-most-amazing-chocolate-chip-cookies/', recipename= "Chocolate Chip Cookies", name='Katelyn', rating='10', comments='Really good recipe, loved')
users = [u1, u2, u3, u4, u5]
"""Builds sample user/note(s) data"""
for user in users:
try:
'''add user to table'''
object = user.create()
print(f"Created new rating {object.recipelink}")
except: # error raised if object nit created
'''fails with bad or duplicate data'''
print(f"Records exist for recipe {user.recipelink}, or error.")
initReview()
def find_by_recipelink(recipelink):
with app.app_context():
user = Review.query.filter_by(_recipelink=recipelink).first()
return user # returns user object
# Check credentials by finding user and verify password
def check_credentials(recipelink, name):
# query email and return user record
user = find_by_recipelink(recipelink)
if user == None:
return False
if (user.is_name(name)):
return True
return False
def find_by_name(name):
with app.app_context():
user = Review.query.filter_by(_name=name).first()
return user # returns user object
# Check credentials by finding user and verify password
def check_credentials(name):
# query email and return user record
user = find_by_recipelink(name)
if user == None:
return False
#check_credentials("indi", "123qwerty")
def create():
# optimize user time to see if uid exists
recipelink = input("Enter your recipe link:")
recipelink = find_by_recipelink(recipelink)
try:
print("Found\n", user.read())
return
except:
pass # keep going
# request value that ensure creating valid object
name = input("Enter YOUR name:")
recipename = input("Enter your recipe name:")
comments = input("Enter any comments you have about this recipe:")
rating = input("Rate this recipe 1-10")
# Initialize User object
user = Review(name=name,
recipelink=recipelink,
recipename=recipename,
comments=comments,
rating=rating
)
# write object to database
with app.app_context():
try:
object = user.create()
print("Created\n", object.read())
except: # if link already exists
print("Someone has already entered this {name}")
create()
def read():
with app.app_context():
table = Review.query.all()
json_ready = [user.read() for user in table] # "List Comprehensions", for each user add user.read() to list
return json_ready
read()
def update():
# optimize user time to see if uid exists
recipelink = input("Enter your recipe link:")
user = find_by_recipelink(recipelink)
if user is None:
print(f"User {recipelink} is not found :(")
return
new_name = input("What is your new name: ")
new_rating = input("What is your new rating: ")
new_comments = input("What are your new comments: ")
with app.app_context():
try:
user.update(new_name, new_rating, new_comments)
print(f"Recipe link, {recipelink}, has been updated with the name, {new_name}, with the rating {new_rating}, and the comments, {new_comments}")
except:
print(f"There was an problem in updating recipe, {recipelink}")
update()
def delete():
# optimize user time to see if uid exists
recipelink = input("Enter your recipe link:")
user = find_by_recipelink(recipelink)
if user is None:
print(f"Recipe, {recipelink} is not found :(")
with app.app_context():
try:
user.delete()
print(f"Recipe, {recipelink} has been deleted.")
except:
print("Enter a recipe link that already exists")
delete()
import sqlite3
database = 'instance/sqlite.db' # this is location of database
def schema():
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL queries
cursor = conn.cursor()
# Fetch results of Schema
results = cursor.execute("PRAGMA table_info('users')").fetchall()
# Print the results
for row in results:
print(row)
# Close the database connection
conn.close()
schema()
import sqlite3
def read():
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL queries
cursor = conn.cursor()
# Execute a SELECT statement to retrieve data from a table
results = cursor.execute('SELECT * FROM favrecipes').fetchall()
# Print the results
if len(results) == 0:
print("Table is empty")
else:
for row in results:
print(row)
# Close the cursor and connection objects
cursor.close()
conn.close()
read()
import sqlite3
def create():
recipelink = input("Enter your recipe link:")
recipename = input("Enter your recipe name:")
name = input("Enter your name:")
rating = input("Enter your rating:")
comments = input("Enter your comments:")
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL commands
cursor = conn.cursor()
try:
# Execute an SQL command to insert data into a table
cursor.execute("INSERT INTO favrecipes (_recipelink, _recipename, _name, _rating, _comments) VALUES (?, ?, ?, ?, ?)", (recipelink, recipename, name, rating, comments))
# Commit the changes to the database
conn.commit()
print(f"A new recipe record {recipelink} has been created")
except sqlite3.Error as error:
print("Error while executing the INSERT:", error)
# Close the cursor and connection objects
cursor.close()
conn.close()
create()
import sqlite3
def update():
recipelink = input("Enter recipe link")
comments = input("Enter new comments")
if len(comments) < 2:
message = "hacked"
comments = '-'
else:
message = "successfully updated"
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL commands
cursor = conn.cursor()
try:
# Execute an SQL command to update data in a table
cursor.execute("UPDATE favrecipes SET _comments = ? WHERE _recipelink = ?", (comments, recipelink))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"Recipe, {recipelink} was not found in the table")
else:
print(f"The row with recipe, {recipelink} the comments has been {message}")
conn.commit()
except sqlite3.Error as error:
print("Error while executing the UPDATE:", error)
# Close the cursor and connection objects
cursor.close()
conn.close()
update()
import sqlite3
def delete():
recipelink = input("Enter recipelink to delete")
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL commands
cursor = conn.cursor()
try:
cursor.execute("DELETE FROM favrecipes WHERE _recipelink = ?", (recipelink,))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"No recipe {recipelink} was found in the table")
else:
# The uid was found in the table and the row was deleted
print(f"The row with recipe, {recipelink} was successfully deleted")
conn.commit()
except sqlite3.Error as error:
print("Error while executing the DELETE:", error)
# Close the cursor and connection objects
cursor.close()
conn.close()
delete()
def menu():
operation = input("Enter: (C)reate (R)ead (U)pdate or (D)elete or (S)chema")
if operation.lower() == 'c':
create()
elif operation.lower() == 'r':
read()
elif operation.lower() == 'u':
update()
elif operation.lower() == 'd':
delete()
elif operation.lower() == 's':
schema()
elif len(operation)==0: # Escape Key
return
else:
print("Please enter c, r, u, or d")
menu() # recursion, repeat menu
try:
menu() # start menu
except:
print("Perform Jupyter 'Run All' prior to starting menu")
#tested R