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.

  • 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?
  1. Flask app object- we've used this in our CPT flask, for example we have used the @app.route decorator to map a URL to a Python function that will be executed when that URL is requested. We also use the init_app() method and can use the request and response objects.
  2. SQLAlchemy object- these are the objects:from flask import Flaskfrom flask_sqlalchemy import SQLAlchemy
"""
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

Define columns, initialization, and CRUD methods for users table in sqlite.db

  • Comment on these items in the class
  • class User purpose
  • db.Model inheritance
  • init method
  • @property, @.setter</li>
  • additional methods
  • </ul>

    all commented in the code

    </div> </div> </div>
    """ 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 Dog class to manage actions in the 'Dogs' 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.) Dog 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 Dog(db.Model): # making template definition of this class on the properties we want this Dog to have 
        # inherent properties into Dog from db model, enables template definition, Dog template can now do database stuff 
        __tablename__ = 'dogs'  # table name is plural, class name is singular
    
        # Define the Dog schema with "vars" from object
        id = db.Column(db.Integer, primary_key=True)
        _image = db.Column(db.String(255), unique=True, nullable=False)
        _link = db.Column(db.String(255), unique=False, nullable=False)
        _name = db.Column(db.String(255), unique=False, nullable=False)
        _uid = db.Column(db.String(255), unique=True, nullable=False)
        _breed = db.Column(db.String(255), unique=False, nullable=False)
        _sex = db.Column(db.String(255), unique=False, nullable=False)
        _dob = db.Column(db.Date)
        _price = db.Column(db.String(255), unique=False, nullable=False)
    
        # constructor of a Dog object, initializes the instance variables within object (dog)
        def __init__(dog, image, link, name, uid, breed, sex, price, dob=datetime.today()):
            # init receives parameters, instantiates an object of our template (template: code), object shows in debugger as a variable with data and methods 
            dog._image = image
            dog._link = link
            dog._name = name    # variables with dog prefix become part of the object, 
            dog._uid = uid
            dog._breed = breed
            dog._sex = sex
            if isinstance(dob, str):  # not a date type     
                dob = datetime=datetime.today()
            dog._dob = dob 
            dog._price = price
    
    # setters and getters retrieve values of attributes inside object, getters receive, setters alter or change the properties and values of objects 
        # a name getter method, extracts name from object
        @property
        def image(dog):
            return dog._image
        
        # a setter function, allows name to be updated after initial object creation
        @image.setter
        def image(dog, image):
            dog._image = image
    
        @property
        def link(dog):
            return dog._link
        
        # a setter function, allows name to be updated after initial object creation
        @link.setter
        def link(dog, link):
            dog._link = link
            
        # name GETTER
        @property
        def name(dog):
            return dog._name
        
        # first name setter 
        @name.setter
        def name(dog, name):
            dog._name = name
        
        # last name getter 
        @property
        def breed(dog):
            return dog._breed
        
        @property
        def uid(dog):
            return dog._uid
    
        # last name setter 
        @uid.setter
        def uid(dog, uid):
            dog._uid = uid
        
        #breed getter 
        @property
        def breed(dog):
            return dog._breed
        
        #breed setter
        @breed.setter
        def breed(dog, breed):
            dog._breed = breed
        
        #hours per week getter    
        @property
        def sex(dog):
            return dog._sex
        
        # sex setter
        @sex.setter
        def sex(dog, sex):
            dog._sex = sex
            
        #coach name getter    
        @property
        def dob(dog):
            dob_string = dog._dob.strftime('%m-%d-%Y')
            return dob_string
        
        # dob should be have verification for type date
        @dob.setter
        def dob(dog, dob):
            dog._dob = dob
    
        @property
        def age(dog):
            today = datetime.today()
            return today.year - dog._dob.year- ((today.month, today.day) < (dog._dob.month, dog._dob.day))
        
        #getter
        @property
        def price(dog):
            return dog._price
        
        #setter
        @price.setter
        def price(dog, price):
            dog._price = price
        # output content using str(object) in human readable form, uses getter
        # output content using json dumps, this is ready for API response
        def __str__(dog):
            return json.dumps(dog.read())
    
        # CRUD create/add a new record to the table
        # returns dog or None on error
        def create(dog): # made methods for them to help them interact with data in our object and perform CRUD 
           # methods help you solve problems with your data 
            try:
                # creates a person object from Dog(db.Model) class, passes initializers
                db.session.add(dog)  # add prepares to persist person object to Dogs table
                db.session.commit()  # SqlAlchemy "unit of work pattern" requires a manual commit
                return dog
            except IntegrityError:
                db.session.remove()
                return None
    
        # CRUD read converts dog to dictionary
        # returns dictionary
        def read(dog):
            return {
                "id": dog.id,
                "image": dog.image,
                "link" : dog.link,
                "name": dog.name,
                "uid": dog.uid,
                "breed": dog.breed,
                "sex": dog.sex,
                "dob": dog.dob,
                "age": dog.age,
                "price": dog.price
            }
    
        # CRUD update: updates Dog name, password, phone
        # returns dog
        def update(dog, image="", link="", name="", uid="", breed="", sex="", price=""):
            """only updates values with length"""
            if len(image) > 0:
                dog.image = image
            if len(link) > 0:
                dog.link = link   
            if len(name) > 0:
                dog.name = name
            if len(uid) > 0:
                dog.uid = uid
            if len(breed) > 0:
                dog.breed = breed
            if len(sex) > 0:
                dog.sex = sex
            if len(price) > 0:
                dog.price = price
            db.session.commit()
            db.session.add(dog) # performs update when id exists\n"
            return dog
    
        # CRUD delete: remove dog
        # None
        def delete(dog):
            db.session.delete(dog)
            db.session.commit()
            return None
        
    

    Initial Data

    Uses SQLALchemy db.create_all() to initialize rows into sqlite.db

    • Comment on how these work?
    1. Create All Tables from db Object:generates database tables based on a defined schema within a database object. Automates the creation of database tables based on a predefined schema, making it easier to set up and manage a database system.2. User Object Constructors: turns input into an object
    2. Try / Except: allows your program to take alternative actions in case an error occurs. Python will first attempt to execute the code in the try statement (code block 1). If no exception occurs, the except statement is skipped and the execution of the try statement is finished.
    """Database Creation and Testing """
    
    
    # Builds working data for testing
    def initDogs():
        with app.app_context():
            """Create database and tables"""
            db.create_all()
            """Tester data for table"""
            u1 = Dog(image='https://do31x39459kz9.cloudfront.net/storage/image/cc7c5dd6a09649e3bf5c6bca96b21daa-1670625496-1670625511-jpg/1024-0-', link='https://haeryny.github.io/teamteam/doginfo/', name='Joe', uid='81729', breed='Labrador Retriever Blend', sex='male', dob=datetime(2022, 2, 11), price='200')
            u2 = Dog(image='https://do31x39459kz9.cloudfront.net/storage/image/672cb9b41e7548f68316d4a328c772d2-1673989499-1673989524-jpg/1024-0-', link='https://haeryny.github.io/teamteam/doginfo/', name='Bean', uid='83792', breed='Shepherd-Rottweiler Blend', sex="male", dob=datetime(2019, 1, 31), price='180')
            u3 = Dog(image='https://do31x39459kz9.cloudfront.net/storage/image/9f57a9ccb04d489c8e0faeb7a6aaecc1-1671755085-1671755107-jpg/1024-0-', link='https://haeryny.github.io/teamteam/doginfo/', name='Harry', uid='80032', breed='Hound-Terrier Blend', sex= "male", dob=datetime(2020, 4, 29), price='160')
            u4 = Dog(image='https://do31x39459kz9.cloudfront.net/storage/image/7a0fd8c5107f469a8b6e3ec6db1bc48a-1671827148-1671827194-jpg/1024-0-', link='https://haeryny.github.io/teamteam/doginfo/', name='Honey', uid='90276', breed='Retriever Blend', sex= "female", dob=datetime(2021, 11, 1), price='200')
            u5 = Dog(image='https://do31x39459kz9.cloudfront.net/storage/image/3b17d9a97b4e41ff984e54467d122820-1670895829-1670895970-jpg/1024-0-', link='https://haeryny.github.io/teamteam/doginfo/', name='George', uid='90277', breed='Retriever Blend', sex= "male", dob=datetime(2021, 11, 1), price='200')
    
    
            dogs = [u1, u2, u3, u4, u5]
    
            """Builds sample Dog/note(s) data"""
            for dog in dogs:
                try:
                    '''add Dog to table'''
                    object = dog.create()
                    print(f"Added new dog {object.name}, with uid {object.uid}")
                except:  # error raised if object nit created
                    '''fails with bad or duplicate data'''
                    print(f"Records exist uid {dog.uid}, or error.")
                    
    initDogs()
    
    Records exist uid 81729, or error.
    Records exist uid 83792, or error.
    Records exist uid 80032, or error.
    Records exist uid 90276, or error.
    Records exist uid 90277, or error.
    

    Check for given Credentials in users table in sqlite.db

    Use of ORM Query object and custom methods to identify user to credentials uid and password

    • Comment on purpose of following
    1. Dog.query.filter_by:query is a method of the SQLAlchemy session object that allows you to query the database for specific data. filter_by(_uid=uid) is a filter that limits the results of the query to only those rows that have a _uid attribute value that matches the uid variable.2. user.password, I changed mine to uid, checking credentials by making sure uid and name matches

    ORM- object relational model

    def find_by_uid(uid):
        with app.app_context():
            dog = Dog.query.filter_by(_uid=uid).first() 
        return dog # returns dog object
    
    # Check credentials by finding dog and verify uid 
    def check_credentials(uid, name):
        # query email and return dog record
        dog = find_by_uid(uid)
        if dog == None:
            return False
        if (dog.name(name)):
            return True
        return False
            
    #check_credentials("indi", "123qwerty")
    

    Create a new User in table in Sqlite.db

    Uses SQLALchemy and custom user.create() method to add row.

    • Comment on purpose of following
    1. user.find_by_uid() and try/except
    2. user = User(...)
    3. user.dob and try/except
    4. user.create() and try/except

    my variables are different, i commentedon these lines though

    def create():
        # optimize user time to see if uid exists
        uid = input("Enter your user id:")
        dog = find_by_uid(uid)
        try:
            print("Found\n", dog.read())
            return
        except:
            pass # keep going
        
        # request value that ensure creating valid object
        image = input("Enter image link:")
        link = input("Enter link to dog info:")
        name = input("Enter dog name:")
        uid = input("Enter UID:")
        breed = input("Enter dog breed:")
        sex = input("Enter dog sex:")
        price = input("Enter dog price:")
        
        # Initialize User object before date
        user = Dog(image=image,
                   link=link,
                    name=name, 
                    uid=uid, 
                    breed=breed,
                    sex=sex,
                    price=price
                    )
        
        # create user.dob, fail with today as dob
        dob = input("Enter dog's date of birth 'YYYY-MM-DD'")
        try:
            user.dob = datetime.strptime(dob, '%Y-%m-%d').date()
        except ValueError:
            user.dob = datetime.today() # setting object equal to dateimte.today()
            print(f"Invalid date {dob} require YYYY-mm-dd, date defaulted to {user.dob}")
               
        # write object to database
        with app.app_context():
            try:
                object = user.create()
                print("Created\n", object.read())
            except:  # error raised if object not created
                print("Unknown error uid {uid}")
            
    create()
    
    ---------------------------------------------------------------------------
    ValueError                                Traceback (most recent call last)
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_20718/551872496.py in <module>
         45             print("Unknown error uid {uid}")
         46 
    ---> 47 create()
    
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_20718/551872496.py in create()
          3     # optimize user time to see if uid exists
          4     uid = input("Enter your user id:")
    ----> 5     dog = find_by_uid(uid)
          6     try:
          7         print("Found\n", dog.read())
    
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_20718/533973387.py in find_by_uid(uid)
          2 def find_by_uid(uid):
          3     with app.app_context():
    ----> 4         dog = Dog.query.filter_by(_uid=uid).first()
          5     return dog # returns dog object
          6 
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/orm/query.py in first(self)
       2750             return self._iter().first()  # type: ignore
       2751         else:
    -> 2752             return self.limit(1)._iter().first()  # type: ignore
       2753 
       2754     def one_or_none(self) -> Optional[_T]:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in first(self)
       1771 
       1772         """
    -> 1773         return self._only_one_row(
       1774             raise_for_second_row=False, raise_for_none=False, scalar=False
       1775         )
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _only_one_row(self, raise_for_second_row, raise_for_none, scalar)
        730         onerow = self._fetchone_impl
        731 
    --> 732         row: Optional[_InterimRowType[Any]] = onerow(hard_close=True)
        733         if row is None:
        734             if raise_for_none:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _fetchone_impl(self, hard_close)
       1658         self, hard_close: bool = False
       1659     ) -> Optional[_InterimRowType[Row[Any]]]:
    -> 1660         return self._real_result._fetchone_impl(hard_close=hard_close)
       1661 
       1662     def _fetchall_impl(self) -> List[_InterimRowType[Row[Any]]]:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _fetchone_impl(self, hard_close)
       2250             self._raise_hard_closed()
       2251 
    -> 2252         row = next(self.iterator, _NO_ROW)
       2253         if row is _NO_ROW:
       2254             self._soft_close(hard=hard_close)
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/orm/loading.py in chunks(size)
        189                     break
        190             else:
    --> 191                 fetch = cursor._raw_all_rows()
        192 
        193             if single_entity:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _raw_all_rows(self)
        519         assert make_row is not None
        520         rows = self._fetchall_impl()
    --> 521         return [make_row(row) for row in rows]
        522 
        523     def _allrows(self) -> List[_R]:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in <listcomp>(.0)
        519         assert make_row is not None
        520         rows = self._fetchall_impl()
    --> 521         return [make_row(row) for row in rows]
        522 
        523     def _allrows(self) -> List[_R]:
    
    lib/sqlalchemy/cyextension/resultproxy.pyx in sqlalchemy.cyextension.resultproxy.BaseRow.__init__()
    
    lib/sqlalchemy/cyextension/processors.pyx in sqlalchemy.cyextension.processors.str_to_date()
    
    ValueError: Invalid isoformat string: ''

    Reading users table in sqlite.db

    Uses SQLALchemy query.all method to read data

    • Comment on purpose of following
    1. User.query.all
    2. json_ready assignment
    # SQLAlchemy extracts all dogs from database, turns each dog into JSON
    def read():
        with app.app_context(): #create a context within which Flask application objects like "Dog" can be accessed.
            table = Dog.query.all() #queries the database to retrieve all rows from the "Dog" table.
        json_ready = [dog.read() for dog in table] # each dog adds dog.read() to list
        #creates a list of JSON-ready data by calling the "read" method on each "dog" object in the "table" list.
        return json_ready
    # returns the list of JSON-ready data as the output of the function.
    
    read()
    
    ---------------------------------------------------------------------------
    ValueError                                Traceback (most recent call last)
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_61102/2924565856.py in <module>
          8 # returns the list of JSON-ready data as the output of the function.
          9 
    ---> 10 read()
    
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_61102/2924565856.py in read()
          2 def read():
          3     with app.app_context(): #create a context within which Flask application objects like "Dog" can be accessed.
    ----> 4         table = Dog.query.all() #queries the database to retrieve all rows from the "Dog" table.
          5     json_ready = [dog.read() for dog in table] # each dog adds dog.read() to list
          6     #creates a list of JSON-ready data by calling the "read" method on each "dog" object in the "table" list.
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/orm/query.py in all(self)
       2695             :meth:`_engine.Result.scalars` - v2 comparable method.
       2696         """
    -> 2697         return self._iter().all()  # type: ignore
       2698 
       2699     @_generative
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in all(self)
       1754 
       1755         """
    -> 1756         return self._allrows()
       1757 
       1758     def __iter__(self) -> Iterator[_R]:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _allrows(self)
        527         make_row = self._row_getter
        528 
    --> 529         rows = self._fetchall_impl()
        530         made_rows: List[_InterimRowType[_R]]
        531         if make_row:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _fetchall_impl(self)
       1661 
       1662     def _fetchall_impl(self) -> List[_InterimRowType[Row[Any]]]:
    -> 1663         return self._real_result._fetchall_impl()
       1664 
       1665     def _fetchmany_impl(
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _fetchall_impl(self)
       2261             self._raise_hard_closed()
       2262         try:
    -> 2263             return list(self.iterator)
       2264         finally:
       2265             self._soft_close()
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/orm/loading.py in chunks(size)
        189                     break
        190             else:
    --> 191                 fetch = cursor._raw_all_rows()
        192 
        193             if single_entity:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in _raw_all_rows(self)
        519         assert make_row is not None
        520         rows = self._fetchall_impl()
    --> 521         return [make_row(row) for row in rows]
        522 
        523     def _allrows(self) -> List[_R]:
    
    ~/opt/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/result.py in <listcomp>(.0)
        519         assert make_row is not None
        520         rows = self._fetchall_impl()
    --> 521         return [make_row(row) for row in rows]
        522 
        523     def _allrows(self) -> List[_R]:
    
    lib/sqlalchemy/cyextension/resultproxy.pyx in sqlalchemy.cyextension.resultproxy.BaseRow.__init__()
    
    lib/sqlalchemy/cyextension/processors.pyx in sqlalchemy.cyextension.processors.str_to_date()
    
    ValueError: Invalid isoformat string: ''

    Hacks

    • Add this Blog to you own Blogging site. In the Blog add notes and observations on each code cell.
    • Add Update functionality to this blog.
    • Add Delete functionality to this blog.

    Delete Functionality Hacks

    import sqlite3
    
    class Dog:
        def __init__(self, db_path):
            self.db_path = db_path
    
        def delete_by_id(self, id):
            # Validate the input
            try:
                id = int(id)
            except ValueError:
                print("Invalid input: ID must be an integer.")
                return
    
            # Connect to the database
            conn = sqlite3.connect(self.db_path)
    
            try:
                # Use a parameterized query to prevent SQL injection
                c = conn.cursor()
                c.execute("DELETE FROM dogs WHERE id=?", (id,))
                conn.commit()
                print(f"Row with ID {id} has been deleted.")
            except sqlite3.Error as e:
                print(f"Error deleting row with ID {id}: {e}")
            finally:
                # Close the connection
                conn.close()
    
    db_path = 'instance/sqlite.db'
    dog = Dog(db_path)
    id = input("Select the ID of the dog you want to remove from our database: ")
    dog.delete_by_id(id)
    
    Row with ID 9 has been deleted.
    

    Update Functionality

    import sqlite3
    
    class Dog:
        def __init__(self, db_path):
            self.db_path = db_path
    def update():
        uid = input("Enter user id to update")
        price = input("Enter updated price")
    
        # 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 dogs SET _price = ? WHERE _uid = ?", (price, uid))
            if cursor.rowcount == 0:
                # The uid was not found in the table
                print(f"No UID {uid} was found in the table")
            else:
                print(f"The row with UID {uid} price has been {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()
    
    ---------------------------------------------------------------------------
    OperationalError                          Traceback (most recent call last)
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_20718/1106125655.py in <module>
         31     conn.close()
         32 
    ---> 33 update()
    
    /var/folders/ry/vf9wstbj0bs5fggn2g7cq48c0000gn/T/ipykernel_20718/1106125655.py in update()
          9 
         10     # Connect to the database file
    ---> 11     conn = sqlite3.connect(database)
         12 
         13     # Create a cursor object to execute SQL commands
    
    OperationalError: unable to open database file
    </div>