2.4a Using Programs with Data
Using data...
""" 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 Score(db.Model):
__tablename__ = 'scores' # 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=False, nullable=False)
_uid = 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, uid):
self._name = name # variables with self prefix become part of the object,
self._uid = uid
# 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 uid(self):
return self._uid
# a setter function, allows name to be updated after initial object creation
@uid.setter
def uid(self, uid):
self._uid = uid
# check if uid parameter matches user id in object, return boolean
def is_uid(self, uid):
return self._uid == uid
@property
def password(self):
return self._password[0:10] + "..." # because of security only show 1st characters
# update password, this is conventional setter
def set_password(self, password):
"""Create a hashed password."""
self._password = generate_password_hash(password, method='sha256')
# 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,
"uid": self.uid,
"dob": self.dob,
"age": self.age,
}
# CRUD update: updates user name, password, phone
# returns self
def update(self, name="", uid="", password=""):
"""only updates values with length"""
if len(name) > 0:
self.name = name
if len(uid) > 0:
self.uid = uid
if len(password) > 0:
self.set_password(password)
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
import sqlite3
def create_connection(db_file):
""" create a database connection to the SQLite database
specified by db_file
:param db_file: database file
:return: Connection object or None
"""
conn = None
try:
conn = sqlite3.connect(db_file)
return conn
except Error as e:
print(e)
return conn
def create_table(conn, create_table_sql):
""" create a table from the create_table_sql statement
:param conn: Connection object
:param create_table_sql: a CREATE TABLE statement
:return:
"""
try:
c = conn.cursor()
c.execute(create_table_sql)
except Error as e:
print(e)
def main():
database = 'instance/class.db'
sql_create_projects_table = """ CREATE TABLE IF NOT EXISTS projects (
id integer PRIMARY KEY,
name text NOT NULL,
begin_date text,
end_date text
); """
sql_create_tasks_table = """CREATE TABLE IF NOT EXISTS tasks (
id integer PRIMARY KEY,
name text NOT NULL,
priority integer,
status_id integer NOT NULL,
project_id integer NOT NULL,
begin_date text NOT NULL,
end_date text NOT NULL,
FOREIGN KEY (project_id) REFERENCES projects (id)
);"""
# create a database connection
conn = create_connection(database)
# create tables
if conn is not None:
# create projects table
create_table(conn, sql_create_projects_table)
# create tasks table
create_table(conn, sql_create_tasks_table)
else:
print("Error! cannot create the database connection.")
if __name__ == '__main__':
main()
def update_schedule():
uid = input("Enter uid of user to be updated: ")
user = find_by_uid(uid)
if not user:
print(f"No user with uid {uid} was found.")
return
name = input("Enter new name: ")
password = input("Enter new password: ")
with app.app_context():
try:
object = user.update(name=name, password=password)
print(f"User with uid --{uid}-- has been updated: {object}")
except: # error raised if object not found
print(f"Error updating user with uid {uid}.")
db = read()
print("---Updated db:---")
print(db)
update_schedule()
def delete_by_uid():
uid = int(input("Enter uid of user to be deleted."))
with app.app_context():
user = db.session.get(User, uid)
if user is not None:
db.session.delete(user)
db.session.commit()
print(f"User with uid --{uid}-- has been deleted.")
users = User.query.all()
print(users)
else:
print(f"No user with uid {uid} was found.")
delete_by_uid()
# SQLAlchemy extracts all users from database, turns each user into JSON
def read():
with app.app_context():
table = User.query.all()
json_ready = [user.read() for schedule in table] # "List Comprehensions", for each user add user.read() to list
return json_ready
read()