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
  2. SQLAlchemy object
    • When using CRUD functions
    • tables of keys and values
"""
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
    • the information collected about the user such as id
  • db.Model inheritance
    • inheriting something from a database.model
    • OOP pillar
    • using those functions to build
  • init method
    • initializes the class object's attributes
  • @property, @.setter
    • getter and setter methods
    • get is read in CRUD
    • set is create/update in CRUD
    </li>
  • additional methods
    • delete
  • </ul> </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 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):
        __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()):
            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
    
        # 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')
    
        # check password parameter versus 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, to avoid unfriendly outcomes
        @property
        def dob(self):
            dob_string = self._dob.strftime('%m-%d-%Y')
            return dob_string
        
        # dob should be have verification for type date
        @dob.setter
        def dob(self, dob):
            if isinstance(dob, str):  # not a date type     
                dob = date=datetime.today()
            self._dob = dob
        
        @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) 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.commit()
            return self
    
        # CRUD delete: remove self
        # None
        def delete(self):
            db.session.delete(self)
            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
    2. User Object Constructors
    3. Try / Except
    """Database Creation and Testing """
    
    
    # Builds working data for testing
    def initUsers():
        with app.app_context():
            """Create database and tables"""
            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]
    
            """Builds sample user/note(s) data"""
            for user in users:
                try:
                    '''add user to table'''
                    object = user.create()
                    print(f"Created new uid {object.uid}")
                except:  # error raised if object nit created
                    '''fails with bad or duplicate data'''
                    print(f"Records exist uid {user.uid}, or error.")
                    
    initUsers()
    
    Unexpected exception formatting exception. Falling back to standard exception
    
    Traceback (most recent call last):
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3256, in _wrap_pool_connect
        The returned object is a proxied version of the DBAPI
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 310, in connect
        for fn, target in events:
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 868, in _checkout
        self.__close(terminate=True)
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 476, in checkout
        def status(self) -> str:
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/impl.py", line 256, in _do_get
        _dialect = _AsyncConnDialect()
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 256, in _create_connection
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 371, in __init__
        self, connection: DBAPIConnection, *, terminate: bool = False
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 666, in __connect
        )
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
        Ultimately we would need to rewrite our "decorator" routine completely
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 661, in __connect
        "fresh",
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/create.py", line 590, in connect
        util.warn_deprecated(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 597, in connect
        )
    sqlite3.OperationalError: unable to open database file
    
    The above exception was the direct cause of the following exception:
    
    Traceback (most recent call last):
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/interactiveshell.py", line 3369, in run_code
        exec(code_obj, self.user_global_ns, self.user_ns)
      File "/tmp/ipykernel_21143/276902621.py", line 30, in <cell line: 30>
        initUsers()
      File "/tmp/ipykernel_21143/276902621.py", line 8, in initUsers
        db.create_all()
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/flask_sqlalchemy/__init__.py", line 1094, in create_all
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/flask_sqlalchemy/__init__.py", line 1086, in _execute_for_all_tables
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/sql/schema.py", line 4864, in create_all
        Defines a composite (one or more column) INDEX.
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3122, in _run_ddl_visitor
        @property
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3038, in begin
        inherits the events of the parent, and new events can be associated
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3210, in connect
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 96, in __init__
        shared among threads using properly synchronized access, it is still
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3289, in raw_connection
        def __init__(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3259, in _wrap_pool_connect
        connection, except that its ``close()`` method will result in the
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2106, in _handle_dbapi_exception_noconnection
        sub_params,
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
        Ultimately we would need to rewrite our "decorator" routine completely
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3256, in _wrap_pool_connect
        The returned object is a proxied version of the DBAPI
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 310, in connect
        for fn, target in events:
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 868, in _checkout
        self.__close(terminate=True)
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 476, in checkout
        def status(self) -> str:
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/impl.py", line 256, in _do_get
        _dialect = _AsyncConnDialect()
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 256, in _create_connection
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 371, in __init__
        self, connection: DBAPIConnection, *, terminate: bool = False
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 666, in __connect
        )
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
        Ultimately we would need to rewrite our "decorator" routine completely
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/pool/base.py", line 661, in __connect
        "fresh",
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/create.py", line 590, in connect
        util.warn_deprecated(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 597, in connect
        )
    sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) unable to open database file
    (Background on this error at: https://sqlalche.me/e/14/e3q8)
    
    During handling of the above exception, another exception occurred:
    
    Traceback (most recent call last):
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/interactiveshell.py", line 1982, in showtraceback
        stb = self.InteractiveTB.structured_traceback(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/ultratb.py", line 1118, in structured_traceback
        return FormattedTB.structured_traceback(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/ultratb.py", line 1012, in structured_traceback
        return VerboseTB.structured_traceback(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/ultratb.py", line 865, in structured_traceback
        formatted_exception = self.format_exception_as_a_whole(etype, evalue, etb, number_of_lines_of_context,
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/ultratb.py", line 818, in format_exception_as_a_whole
        frames.append(self.format_record(r))
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/IPython/core/ultratb.py", line 736, in format_record
        result += ''.join(_format_traceback_lines(frame_info.lines, Colors, self.has_colors, lvals))
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/stack_data/utils.py", line 145, in cached_property_wrapper
        value = obj.__dict__[self.func.__name__] = self.func(obj)
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/stack_data/core.py", line 698, in lines
        pieces = self.included_pieces
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/stack_data/utils.py", line 145, in cached_property_wrapper
        value = obj.__dict__[self.func.__name__] = self.func(obj)
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/stack_data/core.py", line 649, in included_pieces
        pos = scope_pieces.index(self.executing_piece)
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/stack_data/utils.py", line 145, in cached_property_wrapper
        value = obj.__dict__[self.func.__name__] = self.func(obj)
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/stack_data/core.py", line 628, in executing_piece
        return only(
      File "/home/nope1013/anaconda3/lib/python3.9/site-packages/executing/executing.py", line 164, in only
        raise NotOneValueFound('Expected one value, found 0')
    executing.executing.NotOneValueFound: Expected one value, found 0
    

    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. User.query.filter_by
      • find by an attribute
      • check to see if the user has the same id as the input
      • using one object to find the other object
    2. user.password
      • a second check
      • if user is found
      • if password is the same it will be returned true
    def find_by_uid(uid):
        with app.app_context():
            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
        if (user.is_password(password)):
            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
    def create():
        # optimize user time to see if uid exists
        uid = input("Enter your user id:")
        user = find_by_uid(uid)
        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")
        
        # 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'")
        try:
            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.dbo}")
               
        # 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()
    
    Created
     {'id': 7, 'name': 'Bruce Wayne', 'uid': 'bruce', 'dob': '02-20-2020', 'age': 3}
    

    Reading users table in sqlite.db

    Uses SQLALchemy query.all method to read data

    • Comment on purpose of following
    1. User.query.all
      • users are added to table
      • searches for users in TABLE
    2. json_ready assignment
      • turns everybody into JSON
      • this is because JSON is easily readable
      • JSON is universal
    # 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 user in table] # each user adds user.read() to list
        return json_ready
    
    read()
    
    [{'id': 1,
      'name': 'Thomas Edison',
      'uid': 'toby',
      'dob': '02-11-1847',
      'age': 176},
     {'id': 2,
      'name': 'Nikola Tesla',
      'uid': 'niko',
      'dob': '03-15-2023',
      'age': 0},
     {'id': 3,
      'name': 'Alexander Graham Bell',
      'uid': 'lex',
      'dob': '03-15-2023',
      'age': 0},
     {'id': 4,
      'name': 'Eli Whitney',
      'uid': 'whit',
      'dob': '03-15-2023',
      'age': 0},
     {'id': 5,
      'name': 'Indiana Jones',
      'uid': 'indi',
      'dob': '10-21-1920',
      'age': 102},
     {'id': 6,
      'name': 'Marion Ravenwood',
      'uid': 'raven',
      'dob': '10-21-1921',
      'age': 101},
     {'id': 7,
      'name': 'Bruce Wayne',
      'uid': 'bruce',
      'dob': '02-20-2020',
      'age': 3}]
    def update():
        with app.app_context():
            table = User.query.all()
        uid = input("Enter user id to update")
        password = input("Enter updated password")
        if len(password) < 2:
            print("hacked")
            password = 'gothackednewpassword123'
        else:
            print("successfully updated")
    
        # Connect to the database file
    
    
    update()
    
    successfully updated
    
    def delete():
        with app.app_context():
            table = User.query.all()
        uid = input("Enter user id to delete")
     
    delete()
    

    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.
    </div>