Unitdatabaseimplement
testng
"""
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:///database.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)
# this code is supposed to grab your database and open it.
# the flask object creates all of the classes and stores them
# the SQLAlchemy db object essentially serves as a point of interaction for our project, storing all of our variables and information.
""" 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 builds the users in the database
class User(db.Model): # by inheiriting db.Model, we can inheirit the variables and data associated with it, and thus the program can run
__tablename__ = 'facts' # table name is plural, class name is singular
# Define the User schema with "vars" from object
id = db.Column(db.Integer, primary_key=True)
_uid = db.Column(db.String(255), unique=True, nullable=False)
_score = db.Column(db.String(255), unique=False, nullable=False)
# constructor of a User object, initializes the instance variables within object (self)
def __init__(self, uid, score): # this just runs the getters and setters and initializes the database classes/objects
self._score = score # variables with self prefix become part of the object,
self._uid = uid
# a name getter method, extracts name from object
# the getters are the read part of CRUD
@property # this is used to give special functions like setters, getters, and CRUD functions when defining class properties
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 SETTERS are the read part of crud
@property
def uid(self):
return self._uid
# a setter function, allows uid to be updated after initial object creation
# the setters are the create part of CRUD
@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
# score property is returned as string, a string represents date outside object
@property
def score(self):
return self._score
# score setter, verifies date type before it is set or default to today
@score.setter
def score(self, score):
self._score = score
# age is calculated field, age is returned according to date of birth
# 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
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,
"uid": self.uid,
"score": self.score,
}
# CRUD update: updates user name, password, phone
# returns self
def update(self, score="", uid=""):
"""only updates values with length"""
if len(score) > 0:
self.score = score
if len(uid) > 0:
self.uid = uid
db.session.add(self) # performs update when id exists\n",
db.session.commit()
return self
# CRUD delete: remove self
# None
def delete(self):
db.session.delete(self)
db.session.commit()
return None
"""Database Creation and Testing """
# Builds working data for testing
def inits(): # this function is where you would add your users/objects in the data table, and is used to fill table with data.
with app.app_context():
"""Create database and tables"""
db.create_all()
"""Tester data for table"""
u1 = User(score='test', uid='49' )
users = [u1]
"""Builds sample user/note(s) data"""
for user in users: # this creates the user objects
try: # try and except used to prevent duplicates/errors from spilling into the database
'''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.")
inits()
def read():
with app.app_context():
table = User.query.all() # User.query.all() is used to query or gather all of the user objects in the database table and query them for interaction/read function
json_ready = [user.read() for user in table] # "List Comprehensions", for each user add user.read() to list
# list comprehension creates a list using existing lists
return json_ready
read()
def find_by_uid(uid):
with app.app_context():
user = User.query.filter_by(_uid=uid).first() # User.query.filter_by is used to find and single out a object in our database for interaction or password verification in this instance
return user # returns user object
# update functionality
def update():
uid = input("Enter your user id for update:")
user = find_by_uid(uid) # if uid exists, it prints out that user's id exists. shows you which user is deleted
print("Found\n", user.read())
name = input("Enter your name:")
with app.app_context():
user._score = name
db.session.add(user)
db.session.commit()
print("success!")
# delete functionality
def userDelete():
# optimize user time to see if uid exists
uid = input("Enter your user id for deletion:")
user = find_by_uid(uid) # if uid exists, it prints out that user's id exists. shows you which user is deleted
print("Found\n", user.read())
with app.app_context():
("deletion successful")
db.session.delete(user)
db.session.commit()
#userDelete()
update()