Unit 2.4A & 2.4BHacks
"""
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:///tasks.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
''' 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 Task(db.Model):
__tablename__ = 'tasks' # 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.Text, unique=False, nullable=False)
_time = db.Column(db.Integer, unique=False, nullable=False)
_notes = db.Column(db.Text, unique=False, nullable=False)
# constructor of a User object, initializes the instance variables within object (self)
def __init__(self, name, time, notes):
self._name = name
self._time = time
self._notes = notes
# 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 uid from object
@property
def time(self):
return self._time
# a setter function, allows uid to be updated after initial object creation
@time.setter
def time(self, time):
self._uid = time
# a getter method, extracts uid from object
@property
def notes(self):
return self._notes
# a setter function, allows uid to be updated after initial object creation
@notes.setter
def notes(self, notes):
self._notes = notes
# output content using str(object) is in human readable form
# 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,
"time": self.time,
"notes": self.notes,
}
# CRUD update: updates user name, password, phone
# returns self
def update(self, name="", time=0, notes=""):
"""only updates values with length"""
if len(name) > 0:
self.name = name
if time > 0:
self.time = time
if len(notes) > 0:
self.notes = notes
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 initTasks():
with app.app_context():
"""Create database and tables"""
db.create_all()
"""Tester data for table"""
task1 = Task(name="APCSP HW", time=60, notes="pretty time consuming, need a little bit of help")
task2 = Task(name="APWH study", time=30, notes="Need to study for upcoming SAQ exam")
tasks = [task1, task2]
"""Builds sample user/note(s) data"""
for task in tasks:
task.create()
initTasks()
def find_by_task_name(task_name):
with app.app_context():
task = Task.query.filter_by(_name=task_name).first()
return task # returns user object
print(find_by_task_name("APCSP HW"))
def read():
with app.app_context():
table = Task.query.all()
json_ready = [task.read() for task in table] # "List Comprehensions", for each user add user.read() to list
return json_ready
read()
def deleteTask():
task_name = input("Enter name of task to delete: ")
task = find_by_task_name(task_name) # using previous function to locate user by inputted id
with app.app_context():
try:
object = task.delete()
print(f"Task with task name {task_name} has been deleted. ")
db = read()
print(db)
except: # error raised if object not found
(f"No task with the name {task_name} was found.")
deleteTask()
def update_task():
task_name = input("Enter the name of the task you want to update: ")
task = find_by_task_name(task_name)
name = input("Enter new task name:")
time = input("Enter new time:")
notes = input("Enter notes:")
with app.app_context():
try:
object = task.update(_name=name,_time=time, _notes=notes)
print(f"{task_name}-- has been updated: {object} ")
except: # error raised if object not found
(f"The task, {task_name} was not found.")
db = read()
print("Here is the updated DB with all updated tasks:")
print(db)
update_task()
import sqlite3
database = 'instance/tasks.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
# The PRAGMA table_info command prints out the stuff
results = cursor.execute("PRAGMA table_info('tasks')").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 tasks').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():
taskname = input("Enter the task you need to complete: ")
tasktime = input("Enter the amount of time in minutes it will take you: ")
tasknotes = input("Enter any notes about the task or reminders for yourself: ")
# 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 tasks (_name, _time, _notes) VALUES (?, ?, ?)", (taskname, tasktime, tasknotes))
# Commit the changes to the database
conn.commit()
print(f"A new task record {taskname} 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():
name = input("Enter task name to update")
tasktime = input("Enter updated time: ")
if int(tasktime) < 0:
tasktime = "0"
message = "unsuccessfully: DO YOU NOT KNOW SIMPLE THINGS? TIME CAN'T BE NEGATIVE"
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 tasks SET _time = ? WHERE _name = ?", (tasktime, name))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"No taskname {name} was not found in the table")
else:
print(f"The row with taskname {name} 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()
import sqlite3
def delete():
name = input("Enter task name to mark it as complete and 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 tasks WHERE _name = ?", (name,))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"No task name {name} was not found in the table")
else:
# The uid was found in the table and the row was deleted
print(f"The row with task name {name} 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")
Here you can see evidence of me debugging the sqlite code, and I was able to see the connection being established, look at the attributes in the objects, and also see each individual line (each row of the database being accessed) being printed out.
