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

Imports and Flask Objects

Defines and key object creations

  • Comment on where you have observed these working? Provide a defintion of purpose.
    1. Flask app object:I have seen the flask app object being used when we created our databases previously on our flask servers. 2. SQLAlchemy db object: This is also related to the database creations in our flask, talked about above. It is used to work with others data and create our own 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__)
# 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)

Model Definition

""" 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 Movie(db.Model):
    __tablename__ = 'movies'  # 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=True, nullable=False)
    _day = db.Column(db.Date)
    _rate = db.Column(db.String(255), unique=False, nullable=False)
    _comment = 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, rate, comment, day=datetime.today()):
        self._name = name    # variables with self prefix become part of the object, 
        if isinstance(day, str):  # not a date type     
            day = date=datetime.today()
        self._day = day
        self._rate = rate
        self._comment = comment

    # 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
    
    # dob property is returned as string, to avoid unfriendly outcomes
    @property
    def day(self):
        day_string = self._day.strftime('%m-%d-%Y')
        return day_string
    
    # dob should be have verification for type date
    @day.setter
    def day(self, day):
        if isinstance(day, str):  # not a date type     
            day = date=datetime.today()
        self._day = day
    
    # a name getter method, extracts name from object
    @property
    def rate(self):
        return self._rate
    
    # a setter function, allows name to be updated after initial object creation
    @rate.setter
    def rate(self, rate):
        self._rate = rate
    
    @property
    def comment(self):
        return self._comment
    
    # a setter function, allows name to be updated after initial object creation
    @comment.setter
    def comment(self, comment):
        self._comment = comment

    # 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,
            "day": self.day,
            "rate": self.rate,
            "comment": self.comment,
        }

    # CRUD update: updates user name, password, phone
    # returns self
    def update(self, name="", day="", rate="", comment=""):
        """only updates values with length"""
        if len(name) > 0:
            self.name = name

        if isinstance(day, str):  # not a date type     
            day = date=datetime.today()
        else:
            self.day = day

        if len(rate) > 0 and rate >= 0 and rate <= 10:
            self.rate = rate

        if len(comment) > 0:
            self.comment = comment
        
        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

Initial Data

"""Database Creation and Testing """


# Builds working data for testing
def initMovies():
    with app.app_context():
        """Create database and tables"""
        db.create_all()
        """Tester data for table"""
        m1 = Movie(name='Everything Everywhere All At Once', day=datetime(2023, 3, 10), rate=10, comment="New favorite movie!!")
        m2 = Movie(name='Soul', day=datetime(2023, 3, 16), rate=8, comment="Pretty good")
        m3 = Movie(name='Violet Evergarden The Movie', day=datetime(2023, 3, 18), rate=10, comment="Amazing movie to watch after the show.")


        movies = [m1, m2, m3]

        """Builds sample user/note(s) data"""
        for movie in movies:
            try:
                '''add user to table'''
                object = movie.create()
                print(f"Created new movie {movie.name}")
            except:  # error raised if object not created
                '''fails with bad or duplicate data'''
                print(f"Records exist movie {movie.name}, or error.")
                
initMovies()
Created new movie Everything Everywhere All At Once
Created new movie Soul
Created new movie Violet Evergarden The Movie

Check for given Credentials in users table in sqlite.db

def find_by_name(name):
    with app.app_context():
        movie = Movie.query.filter_by(_name=name).first()
    return movie # returns user object

Create a new User in table in Sqlite.db

def create():
    # optimize user time to see if uid exists
    name = input("Enter the name of the movie:")
    movie = find_by_name(name)
    try:
        print("Found\n", movie.read())
        return
    except:
        pass # keep going
    
    # request value that ensure creating valid object
    rate = input("Enter your rating of this movie:")
    comment = input("Enter your comments about this movie:")
    
    # Initialize User object before date
    movie = Movie(name=name, 
                rate=rate,
                comment=comment
                )
    
    # create leader.dob, fail with today as dob
    day = input("Enter the day you watched this movie:")
    try:
        movie.day = datetime.strptime(day, '%Y-%m-%d').date()
    except ValueError:
        movie.day = datetime.today()
        print(f"Invalid date {day} require YYYY-mm-dd, date defaulted to {movie.day}")
    
    # write object to database
    with app.app_context():
        try:
            object = movie.create()
            print("Created\n", object.read())
        except:  # error raised if object not created
            print("Unknown error movie {name}")

create()
Created
 {'id': 4, 'name': 'Top Gun: Maverick', 'day': '06-01-2022', 'rate': '9', 'comment': 'Super thrilling, amazing soundtrack'}

Above, I created a movie to be added to the table

Reading users table in sqlite.db

# SQLAlchemy extracts all movies from database, turns each movie into JSON
def read():
    with app.app_context():
        table = Movie.query.all()
    json_ready = [movie.read() for movie in table] # "List Comprehensions", for each movie add movie.read() to list
    return json_ready

read()
[{'id': 1,
  'name': 'Everything Everywhere All At Once',
  'day': '03-10-2023',
  'rate': '10',
  'comment': 'New favorite movie!!'},
 {'id': 2,
  'name': 'Soul',
  'day': '03-16-2023',
  'rate': '8',
  'comment': 'Pretty good'},
 {'id': 3,
  'name': 'Violet Evergarden The Movie',
  'day': '03-18-2023',
  'rate': '10',
  'comment': 'Amazing movie to watch after the show.'},
 {'id': 4,
  'name': 'Top Gun: Maverick',
  'day': '06-01-2022',
  'rate': '9',
  'comment': 'Super thrilling, amazing soundtrack'}]

You can see the movie that I created, "Top Gun: Maverick", read and displayed, at the bottom of the table

Update Function

Update function to modify/update existing data in the database.

def update():
    # first require name input
    name = input("Enter your movie name:")
    movie = find_by_name(name)
    if movie is None:
        print("Movie with name " + name + " not found.")
        return
    print(movie)
    
    # create new User attributes
    new_name = input("Update your movie name:")
    checkname = find_by_name(new_name)
    if (checkname is not None) and (new_name != name):
        print(f"The movie name {new_name} is a duplicate.")
        return

    new_day = input("Update the date you watched this movie:")
    try:
        dayformat = datetime.strptime(new_day, '%Y-%m-%d').date()
    except ValueError:
        dayformat = datetime.today()
        print(f"Date {new_day} incorrect format. Date set to {dayformat}")
    
    temp_rate = input("Update your movie rating:")
    '''
    if int(temp_rate)<1 or int(temp_rate)>10:
        new_rate = 6
        print(f"Score {temp_rate} invalid. Score set to {new_rate}")
    else:
        new_rate = temp_rate
    '''
    new_comment = input("Update your comments about this movie:")

    # update the leader in the database
    with app.app_context():
        try:
            updated = movie.update(name=new_name, day=dayformat, rate=temp_rate, comment=new_comment)
            print("Updated movie:\n", updated.read())
        except:  # error raised if object not created
            print(f"Updating movie {name} unsucessful.")
        

update()
{"id": 4, "name": "Top Gun: Maverick", "day": "06-01-2022", "rate": "9", "comment": "Super thrilling, amazing soundtrack"}
Updating movie Top Gun: Maverick unsucessful.

Now, I updated the movie "Top Gun: Maverick" to have some different information

Delete Function

Delete function to remove existing data from the database.

def delete():
    # first require uid input
    name = input("Enter the movie to delete:")
    movie = find_by_name(name)
    if movie is None:
        print("Movie with name " + name + " not found.")
        return
    print(movie)

    with app.app_context():
        try:
            movie.delete()
            print("Movie has been deleted successfully.")
        except:
            print("There was an unknown error deleting the given movie.")

delete()
{"id": 4, "name": "Top Gun: Maverick", "day": "01-01-2023", "rate": "8", "comment": "Very fun and thrilling"}
Movie has been deleted successfully.

Lastly, I deleted the movie