-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmodel.py
122 lines (106 loc) · 4.54 KB
/
model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
""" database dependencies to support Users db examples """
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import IntegrityError
from flask_migrate import Migrate
from __init__ import app
# Tutorial: https://www.sqlalchemy.org/library.html#tutorials, try to get into Python shell and follow along
dbURI = 'sqlite:///model/myDB.db'
# Setup properties for the database
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = dbURI
app.config['SECRET_KEY'] = 'SECRET_KEY'
# Create SQLAlchemy engine to support SQLite dialect (sqlite:)
db = SQLAlchemy(app)
Migrate(app, db)
# Define the Users table within the model
# -- Object Relational Mapping (ORM) is the key concept of SQLAlchemy
# -- a.) db.Model is like an inner layer of the onion in ORM
# -- b.) Users 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 Users(db.Model):
# define the Users schema
userID = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(255), unique=False, nullable=False)
email = db.Column(db.String(255), unique=True, nullable=False)
password = db.Column(db.String(255), unique=False, nullable=False)
phone = db.Column(db.String(255), unique=False, nullable=False)
# constructor of a User object, initializes of instance variables within object
def __init__(self, name, email, password, phone):
self.name = name
self.email = email
self.password = password
self.phone = phone
# CRUD create/add a new record to the table
# returns self or None on error
def create(self):
try:
# creates a person object from Users(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 {
"userID": self.userID,
"name": self.name,
"email": self.email,
"password": self.password,
"phone": self.phone,
"query": "by_alc" # This is for fun, a little watermark
}
# CRUD update: updates users name, password, phone
# returns self
def update(self, name, password="", phone=""):
"""only updates values with length"""
if len(name) > 0:
self.name = name
if len(password) > 0:
self.password = password
if len(phone) > 0:
self.phone = phone
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 section"""
def model_tester():
print("--------------------------")
print("Seed Data for Table: users")
print("--------------------------")
db.create_all()
"""Tester data for table"""
u1 = Users(name='Thomas Edison', email='tedison@example.com', password='123toby', phone="1111111111")
u2 = Users(name='Nicholas Tesla', email='ntesla@example.com', password='123niko', phone="1111112222")
u3 = Users(name='Alexander Graham Bell', email='agbell@example.com', password='123lex', phone="1111113333")
u4 = Users(name='Eli Whitney', email='eliw@example.com', password='123whit', phone="1111114444")
u5 = Users(name='John Mortensen', email='jmort1021@gmail.com', password='123qwerty', phone="8587754956")
u6 = Users(name='John Mortensen', email='jmort1021@yahoo.com', password='123qwerty', phone="8587754956")
# U7 intended to fail as duplicate key
u7 = Users(name='John Mortensen', email='jmort1021@yahoo.com', password='123qwerty', phone="8586791294")
table = [u1, u2, u3, u4, u5, u6, u7]
for row in table:
try:
db.session.add(row)
db.session.commit()
except IntegrityError:
db.session.remove()
print(f"Records exist, duplicate email, or error: {row.email}")
def model_printer():
print("------------")
print("Table: users with SQL query")
print("------------")
result = db.session.execute('select * from users')
print(result.keys())
for row in result:
print(row)
if __name__ == "__main__":
model_tester() # builds model of Users
model_printer()