Unit 2.4b Databases Hacks
My hacks fr othe 2.4b lesson, making my own database imperatively
import sqlite3
import csv
import pandas as pd
import tabulate as tb
from io import StringIO as IO
# Grab User input
db_name = input("What name do you want your database to be?")
table_name = input("What name do you want your table to be?")
# Create path for database
database = 'instance/' + db_name + ".db"
# Useful queries to use
create_table_query = f"""CREATE TABLE {table_name} ( id INTEGER PRIMARY KEY AUTOINCREMENT, task TEXT NOT NULL, location TEXT, complete_by DATE, time Text, AMPM Text);"""
check_table_exists_query = f"""SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{table_name}'; """
create_entry_query = f"""INSERT INTO {table_name} (task, location, complete_by, time, AMPM) VALUES (?, ?, ?, ?, ?);"""
check_duplicate_entry_query = f"""SELECT count(*) FROM {table_name} WHERE (task=? AND location=? AND complete_by=? AND time=? AND AMPM=?)"""
drop_table_query = f"""DROP TABLE {table_name};"""
select_all_query = f"""SELECT * FROM {table_name}"""
get_task_query = f"""SELECT task FROM {table_name} WHERE id=?"""
get_location_query = f"""SELECT location FROM {table_name} WHERE id=?"""
get_complete_by_query = f"""SELECT complete_by FROM {table_name} WHERE id=?"""
get_time_query = f"""SELECT time FROM {table_name} WHERE id=?"""
get_ampm_query = f"""SELECT AMPM FROM {table_name} WHERE id=?"""
update_query = f"""UPDATE {table_name} SET task = ?, location = ?, complete_by = ?, time = ?, AMPM = ? WHERE id = ?"""
delete_row_query = f"""DELETE FROM {table_name} WHERE id = ?"""
# Other variables
affirmatives = ["yes", "ye", "y", "ok", "sure"]
def init_table():
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL queries
cursor = conn.cursor()
# Execute table creation query we made earlier
cursor.execute(check_table_exists_query)
if cursor.fetchone()[0]==0 :
cursor.execute(create_table_query)
else:
yn = input("Table already exists, override?")
if yn.lower() in affirmatives:
cursor.execute(drop_table_query)
cursor.execute(create_table_query)
print("Updated Schema, All data wiped")
else:
print("Stopping...")
return
cursor.close()
conn.commit()
conn.close()
init_table()
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('todo')").fetchall()
# Print the results
for row in results:
print(row)
# Close the database connection
conn.close()
schema()
def create(task, location, complete_by, time, am_pm):
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL queries
cursor = conn.cursor()
# Fetch results of Schema
unique = True if cursor.execute(check_duplicate_entry_query, (task, location, complete_by, time, am_pm)).fetchall()[0][0] == 0 else False
if not unique:
yn = input(f"Duplicate entry detected with task: {task}\n location: {location}\n complete by date: {complete_by}\n time: {time}{am_pm}\n Still Add task?")
if yn.lower() not in affirmatives:
print("Not adding duplicate entry.")
return
try:
# Execute an SQL command to insert data into a table, date is in form of 'YYYY-MM-DD'
cursor.execute(create_entry_query, (task, location, complete_by, time, am_pm))
# Commit the changes to the database
conn.commit()
print(f"A new task {task} has been created")
except sqlite3.Error as error:
print("Error while executing the INSERT:", error)
# Close the database connection
cursor.close()
conn.close()
create("Finish USH Poster Project", "Home", "2023-03-17", "7:00", "AM")
create("Start Homework", "Home", "2023-03-18", "4:15", "PM")
#create("Bring out Trash cans", "Home", "2023-03-18", "5:00", "PM")
#create("Cook Dinner", "Home", "2023-03-18", "6:00", "PM")
def import_csv(values):
reader = csv.reader(values.split('|'), delimiter=",")
tasks=[]
locations=[]
complete_by = []
times = []
ampms = []
for entry in reader:
assert len(entry) == 5
tasks.append(entry[0])
locations.append(entry[1])
complete_by.append(entry[2])
times.append(entry[3])
ampms.append(entry[4])
for i in range(len(tasks)):
create(tasks[i], locations[i], complete_by[i], times[i], ampms[i])
import_csv("""Finish Research Literature Review, Home, 2023-03-20, 8:00, PM
Go to Gym, LA Fitness, 2023-03-21, 6:00, PM
Teach Sister Piano, Home, 2023-03-21, 8:00, PM""")
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_all_query).fetchall()
# Print the results
if len(results) == 0:
print("Table is empty")
else:
temp_arr = [[str(j) for j in i] for i in results]
df = pd.DataFrame( data = temp_arr[1:] , columns = temp_arr[0])
df.style.set_properties(**{'text-align': 'left'})
print(df.to_string(index=False))
# Close the cursor and connection objects
cursor.close()
conn.close()
read()
def update():
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL commands
cursor = conn.cursor()
read()
id = input("Enter task id to update")
task = input("Enter updated task (Esc to keep default):") or cursor.execute(get_task_query, (id)).fetchall()[0][0]
location = input("Enter updated location (Esc to keep default):") or cursor.execute(get_location_query, (id)).fetchall()[0][0]
complete_by = input("Enter updated date of completion in YYYY-MM-DD format (Esc to keep default):") or cursor.execute(get_complete_by_query, (id)).fetchall()[0][0]
time = input("Enter updated time (Esc to keep default):") or cursor.execute(get_time_query, (id)).fetchall()[0][0]
ampm = input("Enter AM or PM (Esc to keep default):") or cursor.execute(get_ampm_query, (id)).fetchall()[0][0]
print(id, task, location, complete_by, time, ampm)
# 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_query, (task, location, complete_by, time, ampm, id))
if cursor.rowcount == 0:
# The id was not found in the table
print(f"Task id {id} was not found in the table")
else:
print(f"Task {id} has been updated")
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 delete():
id = input("Enter task number to delete")
# Connect to the database file
conn = sqlite3.connect(database)
# Create a cursor object to execute SQL commands
cursor = conn.cursor()
read()
try:
cursor.execute(delete_row_query, (id))
if cursor.rowcount == 0:
# The uid was not found in the table
print(f"Task {id} was not found in the table")
else:
# The uid was found in the table and the row was deleted
print(f"Task {id} 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():
print("Please select an operation:\n\t 1) Add new entries \n\t 2) Import CSV entries \n\t 3) View Tasks \n\t 4) Update Task \n\t 5) Delete Task \n \t 6) Schema")
option = input()
if option== "1":
print("\nAdding New entry\n")
task = input("Enter updated task:")
location = input("Enter updated location:")
complete_by = input("Enter updated date of completion in YYYY-MM-DD format:")
time = input("Enter updated time:")
ampm = input("Enter AM or PM:")
create(task, location, complete_by, time, ampm)
elif option == "2":
print("\nImporting Entries\n")
to_import = input("Enter raw csv data, seperated by commas, split by new line characters")
import_csv(to_import)
elif option == "3":
print("\nDisplaying Entries\n")
read()
elif option == "4":
print("\nUpdating Entries\n")
update()
elif option == "5":
print("\nDeleting Entries\n")
delete()
elif option == "6":
print("\nDatabase Schema\n")
schema()
else:
print("See you next time!")
return
menu()
menu()
# Finish Research Literature Review, Home, 2023-03-20, 8:00, PM|Go to Gym, LA Fitness, 2023-03-21, 6:00, PM|Teach Sister Piano, Home, 2023-03-21, 8:00, PM