Example - Web Application¶
Let us understand how CRUD operations are performed for Web Applications using Python based frameworks such as Flask. Don’t worry if you do not understand the example completely. By the end of the course, you are supposed to gain fair amount of expertise related to Python in general. In future, we will come up with courses related to Flask.
We typically perform CRUD operations for Web Applications using ORM Frameworks or Libraries.
ORM stands for Object Relational Mapping. Applications typically use objects and databases persist data in the form of tables or relations.
There will not be direct mapping between application objects and database tables or relations. This is where ORM comes into picture.
SQL Alchemy is the most popular Python based ORM library and it is used widely as part of web application frameworks such as Flask.
As part of web application, we typically perform database operations using functions rather than directly running queries.
A good application developer have to be comfortable with both databases as well as functions that comes as part of ORM libraries.
Here are the steps involved to perform database operations using ORM:
Install both database driver such as
psycopg2
as well as ORM such asSQL Alchemy
.Import SQL Alchemy and build database URL.
Create model by inheriting SQL Alchemy Model.
Insert data into the table - one at a time or many at a time.
Query the table to read the data.
%load_ext sql
%env DATABASE_URL=postgresql://itversity_sms_user:itversity!23@localhost:5433/itversity_sms_db
%sql TRUNCATE TABLE users
!pip install psycopg2
!pip install sqlalchemy
import sqlalchemy as db
import psycopg2
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
Table Definition
CREATE TABLE users (
user_id SERIAL PRIMARY KEY,
user_first_name VARCHAR(30) NOT NULL,
user_last_name VARCHAR(30) NOT NULL,
user_email_id VARCHAR(50) NOT NULL,
user_email_validated BOOLEAN DEFAULT FALSE,
user_password VARCHAR(200),
user_role VARCHAR(1) NOT NULL DEFAULT 'U', --U and A
is_active BOOLEAN DEFAULT FALSE,
created_dt DATE DEFAULT CURRENT_DATE,
last_updated_ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
from datetime import datetime, date
class User(Base):
__tablename__ = 'users'
user_id = db.Column(db.Integer, primary_key=True)
user_first_name = db.Column(db.String(30))
user_last_name = db.Column(db.String(30))
user_email_id = db.Column(db.String(50), unique=True)
user_email_validated = db.Column(db.Boolean, default=False)
user_password = db.Column(db.String(10))
user_role = db.Column(db.String(1), default='U')
is_active = db.Column(db.Boolean, default=False)
created_dt = db.Column(db.Date, default=date.today())
last_updated_ts = db.Column(db.DateTime, default=datetime.now())
def __repr__(self):
return """<User(user_id=%s, user_first_name='%s', user_last_name='%s',
user_email_id='%s', user_email_validated='%s', user_password='%s',
user_role='%s', is_active='%s', created_dt='%s', last_updated_ts='%s'
)>""" % (
self.user_id, self.user_first_name, self.user_last_name,
self.user_email_id, self.user_email_validated, self.user_password,
self.user_role, self.is_active, self.created_dt, self.last_updated_ts
)
engine = db.create_engine('postgresql://itversity_sms_user:itversity!23@localhost:5433/itversity_sms_db')
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
user = User(
user_first_name='Gordan',
user_last_name='Bradock',
user_email_id='gbradock0@barnesandnoble.com',
user_role='A',
created_dt='2020-01-10'
)
print(user)
session.add(user)
session.commit()
session.query(User).first()
%sql SELECT * FROM users
users = [
('Tobe', 'Lyness', 'tlyness1@paginegialle.it', '2020-02-10'),
('Addie', 'Mesias', 'amesias2@twitpic.com', '2020-03-05'),
('Corene', 'Kohrsen', 'ckohrsen3@buzzfeed.com', '2020-04-15'),
('Darill', 'Halsall', 'dhalsall4@intel.com', '2020-10-10')
]
user_objects = map(
lambda user: User(user_first_name=user[0], user_last_name=user[1], user_email_id=user[2], created_dt=user[3]),
users
)
session.add_all(user_objects)
session.commit()
%sql SELECT * FROM users
for user in session.query(User).all():
print(user)
for user in session.query(User).all():
print(user.user_email_id)
session.close()