Unit 2.4a Using Programs with Data, SQLAlchemy
Using Programs with Data is focused on SQL and database actions. Part A focuses on SQLAlchemy and an OOP programming style,
Database and SQLAlchemy
In this blog we will explore using programs with data, focused on Databases. We will use SQLite Database to learn more about using Programs with Data. Use Debugging through these examples to examine Objects created in Code.
-
College Board talks about ideas like
- Program Usage. "iterative and interactive way when processing information"
- Managing Data. "classifying data are part of the process in using programs", "data files in a Table"
- Insight "insight and knowledge can be obtained from ... digitally represented information"
- Filter systems. 'tools for finding information and recognizing patterns"
- Application. "the preserve has two databases", "an employee wants to count the number of book"
-
PBL, Databases, Iterative/OOP
- Iterative. Refers to a sequence of instructions or code being repeated until a specific end result is achieved
- OOP. A computer programming model that organizes software design around data, or objects, rather than functions and logic
- SQL. Structured Query Language, abbreviated as SQL, is a language used in programming, managing, and structuring data
"""
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__) # initializes the Flask object we are going to use
# 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()
# initializes the database with 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 User(db.Model):
# 1) this is a User object that is a class which is used to define properties of the user with variables
# 2) db.Model is inherited towards the specific user on which the instance that it is created, storing the data needed
__tablename__ = 'users' # 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=True, nullable=False)
_password = db.Column(db.String(255), unique=False, nullable=False)
_dob = db.Column(db.Date)
# constructor of a User object, initializes the instance variables within object (self)
def __init__(self, name, uid, password="123qwerty", dob=datetime.today()):
# 3) this method is used to initialize the properties and state of the instance which is running, in this case setting the variables that will be used to store the data into the database
self._name = name # variables with self prefix become part of the object,
self._uid = uid
self.set_password(password)
if isinstance(dob, str): # not a date type
dob = date=datetime.today()
self._dob = dob
# 4) @property defines a property for the database model which is accessed like an attribute, but is actually a method
# 5) @<column>.setter is a method that defines the value of the database column
# 4-5) both of these are used to initialize the data that will be placed into the database
# 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 uid(self):
return self._uid
# a setter function, allows uid 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 method used for setter
def set_password(self, password):
"""Create a hashed password."""
self._password = generate_password_hash(password, method='sha256')
# check password parameter against stored/encrypted password
def is_password(self, password):
"""Check against hashed password."""
result = check_password_hash(self._password, password)
return result
# dob property is returned as string, a string represents date outside object
@property
def dob(self):
dob_string = self._dob.strftime('%m-%d-%Y')
return dob_string
# dob setter, verifies date type before it is set or default to today
@dob.setter
def dob(self, dob):
if isinstance(dob, str): # not a date type
dob = date=datetime.today()
self._dob = dob
# age is calculated field, age is returned according to date of birth
@property
def age(self):
today = datetime.today()
return today.year - self._dob.year - ((today.month, today.day) < (self._dob.month, self._dob.day))
# 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
# 6) these methods are used to add, update, delete, and remove data from the database of stored values
# 6) create: is used to make a new instance of the database and add information to it
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
# 6) read: is used to retrieve specified instances from the database and send them to the frontend
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
# 6) update: is used to modify existing information in the database
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.commit()
return self
# CRUD delete: remove self
# None
# 6) delete: is used to remove existing information from the database
def delete(self):
db.session.delete(self)
db.session.commit()
return None
"""Database Creation and Testing """
# Builds working data for testing
def initUsers():
with app.app_context():
"""Create database and tables"""
# 1) this adds all of the info into the database through the create function, adding in info based on the setters and getters of the User object defined
db.create_all()
"""Tester data for table"""
u1 = User(name='Thomas Edison', uid='toby', password='123toby', dob=datetime(1847, 2, 11))
u2 = User(name='Nikola Tesla', uid='niko', password='123niko')
u3 = User(name='Alexander Graham Bell', uid='lex', password='123lex')
u4 = User(name='Eli Whitney', uid='whit', password='123whit')
u5 = User(name='Indiana Jones', uid='indi', dob=datetime(1920, 10, 21))
u6 = User(name='Marion Ravenwood', uid='raven', dob=datetime(1921, 10, 21))
users = [u1, u2, u3, u4, u5, u6]
# 2) this creates a new instance for each user, creating the user data into an object and adding it into the database
"""Builds sample user/note(s) data"""
for user in users:
# 3) try means that this block of code will be run but may result in some exceptions
try:
'''add user to table'''
object = user.create()
print(f"Created new uid {object.uid}")
# 4) except means that once everything is tried and if exceptions are met, this code will run, which in this case will result in an error message
except: # error raised if object nit created
'''fails with bad or duplicate data'''
print(f"Records exist uid {user.uid}, or error.")
initUsers()
def find_by_uid(uid):
with app.app_context():
# 1) this searches through all of the data objects based on the specifies variable, which in this case is the user id
# in this case, this program code is used to search for the user id that is entered
user = User.query.filter_by(_uid=uid).first()
return user # returns user object
# Check credentials by finding user and verify password
def check_credentials(uid, password):
# query email and return user record
user = find_by_uid(uid)
if user == None:
return False
# 2) this checks if the password that is entered by the user and that matches with the object that they selected
# is the password matches, then True is returned
if (user.is_password(password)):
return True
return False
#check_credentials("indi", "123qwerty")
def create():
# optimize user time to see if uid exists
uid = input("Enter your user id:")
# 1) this searches through the objects created in the database based on the request that is given by searching for a certain user id
user = find_by_uid(uid)
# 1) this tries to see if the user id matches up with a uid in the database and if it does, then it returns True, if it doesn't then it keeps searching
try:
print("Found\n", user.read())
return
except:
pass # keep going
# request value that ensure creating valid object
name = input("Enter your name:")
password = input("Enter your password")
# 2) this sets the User object and the variables that store the data inside the object
# Initialize User object before date
user = User(name=name,
uid=uid,
password=password
)
# create user.dob, fail with today as dob
dob = input("Enter your date of birth 'YYYY-MM-DD'")
# 3) tries to format the date that is inputted by the user and then puts it in the database
# if the date isn't formatted correctly, an error is returned to the user
try:
# 3) this defines the date that is entered to the user object and formats it into the correct sequence
user.dob = datetime.strptime(dob, '%Y-%m-%d').date()
except ValueError:
user.dob = datetime.today()
print(f"Invalid date {dob} require YYYY-mm-dd, date defaulted to {user.dob}")
# write object to database
with app.app_context():
# 4) this attempts to create a user based on the data given and if the data doesn't match up the data is rejected and an error is given
try:
# 4) the user is created and assigned to the variable `object`
object = user.create()
print("Created\n", object.read())
except: # error raised if object not created
print("Unknown error uid {uid}")
create()
# SQLAlchemy extracts all users from database, turns each user into JSON
def read():
with app.app_context():
# 1) this goes through the entire user object created and reads all of the data that is in the database, outputting it
table = User.query.all()
# 2) this assigns the data into a JSON format to be sent to the frontend without additional formatting by the program
# google list comprehension is the way that lists are created in Python and manipulated for future use within Google's codebase
json_ready = [user.read() for user in table] # "List Comprehensions", for each user add user.read() to list
return json_ready
read()
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
dbURI = 'sqlite:///instance/cars.db'
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SQLALCHEMY_DATABASE_URI"] = dbURI
app.config["SECRET_KEY"] = "SECRET_KEY"
db = SQLAlchemy(app)
db.init_app(app)
from sqlalchemy import Column, Integer, String
class Car(db.Model):
# defining table
__tablename__ = "car"
id = Column(Integer, primary_key=True)
_manufacturer = Column(Integer, nullable=False)
_model = Column(String(255), nullable=False)
_price = Column(Integer, nullable=False)
# initialization
def __init__(self, manufacturer, model, price):
self._manufacturer = manufacturer
self._model = model
self._price = price
def __repr__(self):
return "<Timer(id='%s', manufacturer='%s', model='%s', price='%s')>" % (
self.id,
self.manufacturer,
self.model,
self.price
)
@property
def manufacturer(self):
return self._manufacturer
@manufacturer.setter
def text(self, manufacturer):
self._text = manufacturer
@property
def model(self):
return self._model
@model.setter
def model(self, model):
self._text = model
@property
def price(self):
return self._price
@price.setter
def price(self, value):
self._price = value
def to_dict(self):
return {"id": self.id, "manufacturer": self.manufacturer, "model": self.model, "price": self.price}
def init_cars():
car1 = Car(manufacturer="lamborghini", model="Superveloce Jota", price=517770)
car2 = Car(manufacturer="ferrari", model=" SF90 Stradale", price=528765)
car3 = Car(manufacturer="nissan", model="GT-R", price=116040)
car4 = Car(manufacturer="mazda", model="MX-5", price=29115)
car5 = Car(manufacturer="bmw", model="M5", price=105695)
db.session.add(car1)
db.session.add(car2)
db.session.add(car3)
db.session.add(car4)
db.session.add(car5)
db.session.commit()
db.create_all()
init_cars()
from flask import Blueprint, request
from flask_restful import Api, Resource, reqparse
car_bp = Blueprint("car", __name__)
car_api = Api(car_bp)
class CarAPI(Resource):
def get(self):
id = request.args.get("id")
car = db.session.query(Car).get(id)
if car:
return car.to_dict()
return {"message": "car not found"}, 404
def post(self):
parser = reqparse.RequestParser()
parser.add_argument("manufacturer", required=True, type=str)
parser.add_argument("model", required=False, type=str)
parser.add_argument("price", required=False, type=int)
args = parser.parse_args()
car = Car(args["manufacturer"], args["model"], args["price"])
try:
db.session.add(car)
db.session.commit()
return car.to_dict(), 201
except Exception as e:
db.session.rollback()
return {"message": f"server error: {e}"}, 500
def put(self):
parser = reqparse.RequestParser()
parser.add_argument("id", required=True, type=int)
args = parser.parse_args()
try:
car = db.session.query(Car).get(args["id"])
if car:
car.started = False
db.session.commit()
else:
return {"message": "car not found"}, 404
except Exception as e:
db.session.rollback()
return {"message": f"server error: {e}"}, 500
def delete(self):
parser = reqparse.RequestParser()
parser.add_argument("id", required=True, type=int)
args = parser.parse_args()
try:
car = db.session.query(Car).get(args["id"])
if car:
db.session.delete(car)
db.session.commit()
return car.to_dict()
else:
return {"message": "car not found"}, 404
except Exception as e:
db.session.rollback()
return {"message": f"server error: {e}"}, 500
class CarListAPI(Resource):
def get(self):
cars = db.session.query(Car).all()
return [car.to_dict() for car in cars]
def delete(self):
try:
db.session.query(Car).delete()
db.session.commit()
return []
except Exception as e:
db.session.rollback()
return {"message": f"server error: {e}"}, 500