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?
- 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.
- 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?
- 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
- 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()
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
- 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")
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()
# 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()
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)
import sqlite3 message = "successfully updated" database = 'instance/sqlite.db' 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()