Unit 2.4 My Own Database blog
recommend!
- Imports and Flask Objects
- Model Definition
- Initial Data
- Check for given Credentials in users table in sqlite.db
- Create a new User in table in Sqlite.db
- Reading users table in sqlite.db
- Update
- Delete
- Create
- CRUD menu
"""
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
class Flavor(db.Model):
__tablename__ = 'flavor' # table name is plural, class name is singular
# Define the User schema with "vars" from object
id = db.Column(db.Integer, primary_key=True)
_name = db.Column(db.String(255), unique=True, nullable=False)
_flavor = db.Column(db.String(255), unique=False, nullable=False)
_size = db.Column(db.String(255), unique=False, nullable=False)
_recommend = db.Column(db.String(255), unique=False, nullable=False)
# constructor of a User object, initializes the instance variables within object (self)
def __init__(self, name, flavor, size, recommend):
self._name = name # variables with self prefix become part of the object,
self._flavor = flavor
self._size = size
self._recommend = recommend
# a name getter method, extracts name from object
@property
def name(self):
return self._name
# a setter function, allows name to be updated after initial object creation
@name.setter
def name(self, name):
self._name = name
# a getter method, extracts email from object
@property
def flavor(self):
return self._flavor
# a setter function, allows name to be updated after initial object creation
@flavor.setter
def flavor(self, flavor):
self._flavor = flavor
# check if uid parameter matches user id in object, return boolean
#def is_flavor(self, flavor):
# return self._flavor == flavor
@property
def size(self):
return self._size
# a setter function, allows name to be updated after initial object creation
@size.setter
def size(self, size):
self._size = size
@property
def recommend(self):
return self._recommend
# a setter function, allows name to be updated after initial object creation
@name.setter
def name(self, recommend):
self._recommend = recommend
# output content using str(object) in human readable form, uses getter
# output content using json dumps, this is ready for API response
def __str__(self):
return json.dumps(self.read())
# CRUD create/add a new record to the table
# returns self or None on error
def create(self):
try:
# creates a person object from User(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,
"name": self.name,
"flavor": self.flavor,
"size": self.size,
"recommend": self.recommend,
}
# CRUD update: updates user name, password, phone
# returns self
def update(self, name="", flavor="", size="", recommend=""):
"""only updates values with length"""
if len(name) > 0:
self.name = name
if len(flavor) > 0:
self.flavor = flavor
if len(size) > 0:
self.size = size
if len(recommend) > 0:
self.recommend = recommend
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 initFlavor():
with app.app_context():
"""Create database and tables"""
db.create_all()
"""Tester data for table"""
i1 = Flavor(name='Sherlock', flavor='chocolate', size='small', recommend='yes')
i2 = Flavor(name='Dude', flavor='vanilla', size='medium', recommend='yes')
i3 = Flavor(name='Bunny', flavor='mint', size='large', recommend='yes')
i4 = Flavor(name='Lion', flavor='mango', size='small', recommend='yes')
i5 = Flavor(name='Penguin', flavor='rocky road', size='medium', recommend='yes')
i6 = Flavor(name='Turtle', flavor='coconut', size='small', recommend='yes')
flavor = [i1, i2, i3, i4, i5, i6]
"""Builds sample user/note(s) data"""
for flavor in flavor:
try:
'''add flavor to table'''
object = flavor.create()
print(f"Created new name {object.name}")
except: # error raised if object nit created
'''fails with bad or duplicate data'''
print(f"Records exist name {flavor.flavor}, or error.")
initFlavor()
def find_by_flavor(flavor):
with app.app_context():
flavor = Flavor.query.filter_by(_flavor=flavor).first()
return flavor # returns user object
# Check credentials by finding user and verify password
def check_credentials(flavor):
# query email and return user record
flavor = find_by_flavor(flavor)
if flavor == None:
return False
if (flavor.is_flavor(flavor)):
return True
return False
#check_credentials("indi", "123qwerty")
def create():
flavor = input("Enter your flavor:")
#flavor = find_by_flavor(flavor)
#try:
# print("Found\n", flavor.read())
# return
#except:
# pass # keep going
# request value that ensure creating valid object
name = input("Enter your name:")
size = input("Enter your size:")
recommend = input("Do you recommend this?")
# password = input("Enter your password")
# Initialize User object before date
flavor = Flavor(name=name,
flavor=flavor,
size=size,
recommend=recommend
)
# write object to database
with app.app_context():
try:
object = flavor.create()
print("Created\n", object.name)
except: # error raised if object not created
print("Unknown error name {name}")
#create()
def read():
with app.app_context():
table = Flavor.query.all()
json_ready = [flavor.read() for flavor 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
flavor = input("Enter the flavor you tried and want to re-review")
user = find_by_flavor(flavor)
if flavor is None:
print(f"User {flavor} is not found :(")
return
new_name = input("What is your name?")
new_size = input("What size did you order?")
new_recommend = input("Do you still recommend this product?")
with app.app_context():
try:
user.update(new_name, new_size, new_recommend)
print(f"The flavor, {flavor}, has been updated by {new_name}, with the size {new_size}, and they said: {new_recommend} they do recommend it.")
except:
print(f"There was an problem in updating the review, {flavor}")
#update()
def delete():
# optimize user time to see if uid exists
flavor = input("Enter the flavor you review:")
user = find_by_flavor(flavor)
if user is None:
print(f"Flavor, {flavor} is not found :(")
with app.app_context():
try:
user.delete()
print(f"Flavor, {flavor} has been deleted.")
except:
print("Enter a flavor 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 flavor').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
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 create():
flavor = input("Enter your flavor")
#name = input("Enter your recipe name:")
name = input("Enter your name:")
size = input("Enter your size:")
recommend = input("Do you recommend this flavor?")
# 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 flavor (_flavor, _name, _size, _recommend) VALUES (?, ?, ?, ?)", (flavor, name, size, recommend))
# Commit the changes to the database
conn.commit()
print(f"A new recipe record {flavor} 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()
# def update():
# flavor = input("Enter flavor to update")
# recommend = input("Do you still recommend it?")
# # 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 flavor SET _recommend = ? WHERE _flavor = ?", (recommend, flavor))
# if cursor.rowcount == 0:
# # The uid was not found in the table
# print(f"Flavor, {flavor} was not found in the table")
# else:
# print(f"The row with flavor, {flavor} the comments has been {recommend}")
# 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():
flavor = input("Enter flavor 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 flavor WHERE _flavor = ?", (flavor,))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"No flavor {flavor} was found in the table")
else:
# The uid was found in the table and the row was deleted
print(f"The row with flavor, {flavor} 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()
import sqlite3
def update():
flavor = input("Enter flavor")
recommend = input("Enter recommendation")
# if len(password) < 2:
# message = "hacked"
# password = 'gothackednewpassword123'
# 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 flavor SET _recommend = ? WHERE _flavor = ?", (recommend, flavor))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"No flavor {flavor} was not found in the table")
else:
print(f"The row with flavor {flavor} the password has been {recommend}")
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()
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