Unit 2.4b Using Programs with Data, SQL Practice/test NB for hacks
Using Programs with Data is focused on SQL and database actions. Part B focuses on learning SQL commands, connections, and curses using an Imperative programming style,
Hacks
- Add this Blog to you own Blogging site. In the Blog add notes and observations on each code cell.
- In this implementation, do you see procedural abstraction?
Yes. In the code above, the schema, read, create, update, and delete methods are defined and used later in the program for the menu. This is an example of procedural abstraction because it provides a more modular approach to programming the menu, as we can call the defined methods based on what option the user chooses instead of defining them separately each time the user selects an option.
- In 2.4a or 2.4b lecture
- Do you see data abstraction? Complement this with Debugging example.
- Use Imperative or OOP style to Create a new Table or do something that applies to your CPT project.
Below is a table that stores data about employees such as employee id number, name, hours worked, and salary using OOP.
I added a menu interface at the bottom, which allows the employee to execute methods defined in the code below.
Reference... sqlite documentation
"""
These imports define the key objects
"""
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
""" database dependencies to support sqlite examples """
import json
from sqlalchemy.exc import IntegrityError
"""
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)
# Athlete table
class Athlete(db.Model):
__tablename__ = 'athletes'
# Define the User schema with "vars" from object
id = db.Column(db.Integer, primary_key=True)
_eid = db.Column(db.String(255), unique=False, nullable=False)
_bench = db.Column(db.String(255), unique=False, nullable=False)
_squat = db.Column(db.String(255), unique=False, nullable=False)
_pullup = db.Column(db.String(255), unique=False, nullable=False)
def __init__(self, eid, bench, squat, pullup):
self._eid = eid
self._bench = bench
self._squat = squat
self._pullup = pullup
@property
def eid(self):
return self._eid
@eid.setter
def eid(self, eid):
self._eid = eid
@property
def bench(self):
return self._bench
@bench.setter
def bench(self, bench):
self._bench = bench
@property
def squat(self):
return self._squat
@squat.setter
def squat(self, squat):
self._squat = squat
@property
def pullup(self):
return self._pullup
@pullup.setter
def pullup(self, pullup):
self._pullup = pullup
# output content using json dumps, this is ready for API response
def __str__(self):
return json.dumps(self.read())
def create(self):
try:
db.session.add(self)
db.session.commit()
return self
except IntegrityError:
db.session.remove()
return None
# CRUD read converts self to dictionary
# returns dictionary
def read(self):
return {
"eid": self.eid,
"bench": self.bench,
"squat": self.squat,
"pullup": self.pullup,
}
# CRUD update: updates user name, password, phone
# returns self
def update(self, eid="", bench="", squat="", pullup=""):
"""only updates values with length"""
if len(eid) > 0:
self.eid = eid
if len(bench) > 0:
self.bench = bench
if len(squat) > 0:
self.squat(squat)
if len(pullup) > 0:
self.pullup(pullup)
db.session.add(self)
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 initAthlete():
with app.app_context():
"""Create database and tables"""
db.create_all()
"""Tester data for table"""
# Input data for objects which are made from the template defined by 'User'
e1 = Athlete(eid = 1, bench = '225', squat='400', pullup='12')
e2 = Athlete(eid = 2, bench = '135', squat='225', pullup='22')
e3 = Athlete(eid = 3, bench = '180', squat='300', pullup='15')
athletes = [e1, e2, e3]
"""Builds sample player/note(s) data"""
for e in athletes:
try:
'''add user to table'''
object = e.create()
print(f"Created new athlete {object.eid}")
except: # error raised if object not created
'''fails with bad or duplicate data'''
print(f"Records exist name {e.eid}, or error.")
initAthlete()
def find_by_eid(eid):
with app.app_context():
# ORM allows us to do query methods on our data
# only returns the match to the first match to the database
athlete = Athlete.query.filter_by(_eid=eid)
return athlete # returns user object
def check_credentials(eid):
# query email and return user record
Athlete = find_by_eid(eid)
if Athlete == None:
return False
if (Athlete.is_eid(eid)):
return True
return False
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('employees')").fetchall()
# Print the results
for row in results:
print(row)
# Close the database connection
conn.close()
def create():
eid = input("Enter new athlete number (Any Number)")
bench = input("Enter your max bench")
squat = input("Enter your max squat")
pullup = input('Enter your max pullup')
# 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 athletes (_eid, _bench, _squat, _pullup) VALUES (?, ?, ?, ?)", (eid, bench, squat, pullup))
# Commit the changes to the database
conn.commit()
print(f"A new user record {eid} 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()
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 athletes').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()
def delete():
eid = input("Enter athlete id 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 athletes WHERE _eid = ?", (eid,))
if cursor.rowcount == 0:
# The eid was not found in the table
print(f"No eid {eid} was not found in the table")
else:
# The eid was found in the table and the row was deleted
print(f"The row with eid {eid} 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()
import sqlite3
def update():
eid = input("Enter updated Athlete ID: ")
if int(eid) < 0:
eid = "0"
message = "unsuccessfully: DO YOU NOT KNOW SIMPLE THINGS? ID CAN'T BE NEGATIVE"
else:
message = "successfully updated"
bench = input("Enter bench press update")
squat = input("Enter squat update")
pullup = input("Enter pullup update")
# 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 athletes SET _bench = ? SET _squat = ? SET _pullup = ? WHERE _eid = ?", (bench, squat, pullup))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"No Athlete {eid} was not found in the table")
else:
print(f"The row with Athlete {eid} the time has been udpated {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()
def update_athlete():
eid = input("Enter the ID of the Athlete you want to update: ")
Athlete = find_by_eid(eid)
bench = input("Enter new Bench")
with app.app_context():
try:
object = Athlete.update(eid=eid, bench=bench)
print(f"{eid}-- has been updated: {object} ")
except: # error raised if object not found
(f"The task, {eid} was not found.")
db = read()
print("Here is the updated DB with all updated tasks:")
print(db)
update_athlete()
def updateA():
eid = input("Enter the Athete ID to update: ")
Athlete = find_by_eid(eid)
# Request new values for the QB attributes
bench = input("Enter their new Bench Press")
squat = input("Enter their new Squat ")
pullup = input("Enter their new pullup")
if Athlete is not None:
with app.app_context():
Athlete.update(bench = bench, squat=squat, pullup=pullup)
print("Updated Athlete", eid)
else:
print("error")
updateA()
options = [
('c', 'Create'),
('r', 'Read'),
('u', 'Update'),
('d', 'Delete'),
('s', 'Schema')
]
# Define the menu function
def menu():
# Display the menu options
print('Select an option:')
for option in options:
print(f'({option[0]}) {option[1]}')
# Get the user's choice
operation = input("See the output below and choose an option, or press 'enter' to exit.").lower()
# Find the selected option in the options list
selected_option = None
for option in options:
if operation == option[0]:
selected_option = option
break
# Call the corresponding function for the selected option
if selected_option:
if selected_option[0] == 'c':
create()
elif selected_option[0] == 'r':
read()
elif selected_option[0] == 'u':
update_athlete()
elif selected_option[0] == 'd':
delete()
elif selected_option[0] == 's':
schema()
elif operation == '':
return
else:
print('Invalid option')
# Repeat the menu
menu()
menu()