Python Flask - SQLite connection using SQLAlchemy

 In the previous Python Flask-MySQL connection using SQLAlchemy, we learned how to integrate MySQL in Flask. This time, I will use SQLite, which can be used lightly. Basically, there is no difference in how to use SQLAlchemy even if you use MySQL or SQLite. This time, I will implement it by paying more attention to the UI. I do a lot of work in the call center. Recently, a customer requested an ARS system for callback processing. To implement the system at low cost, H/W used Odroid XU4, SBC (Single Board Computer). I installed the IVR system here, and decided to use the Flask web server and SQLite for callback management.

I learned a lot from Parwiz's Flask CRUD Application with SQLAlchemy and used it to write this article.

Prerequisites

It is helpful to read the following article in advance.

Install the necessary software by referring to the link above.

And install flask_wtf with pip3 command. 

root@ubuntusrv:/usr/local/src/study/flask_sqlite# pip3 install flask_wtf
Collecting flask_wtf
  Downloading Flask_WTF-0.14.3-py2.py3-none-any.whl (13 kB)
Requirement already satisfied: WTForms in /usr/local/lib/python3.8/dist-packages (from flask_wtf) (2.3.3)
Requirement already satisfied: Flask in /usr/local/lib/python3.8/dist-packages (from flask_wtf) (1.1.2)
Requirement already satisfied: itsdangerous in /usr/local/lib/python3.8/dist-packages (from flask_wtf) (1.1.0)
Requirement already satisfied: MarkupSafe in /usr/lib/python3/dist-packages (from WTForms->flask_wtf) (1.1.0)
Requirement already satisfied: click>=5.1 in /usr/lib/python3/dist-packages (from Flask->flask_wtf) (7.0)
Requirement already satisfied: Jinja2>=2.10.1 in /usr/lib/python3/dist-packages (from Flask->flask_wtf) (2.10.1)
Requirement already satisfied: Werkzeug>=0.15 in /usr/local/lib/python3.8/dist-packages (from Flask->flask_wtf) (1.0.1)
Installing collected packages: flask-wtf
Successfully installed flask-wtf-0.14.3


Create Callback Table using SQLAlchemy

First, in the IVR system, create a table where customers can leave their contact phone numbers. This table will implement a screen that can be viewed and modified later in Flask.

# create_tables.py
from sqlalchemy import create_engine, ForeignKey, Index
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:////usr/local/src/study/flask_sqlite/callback.db')
Base = declarative_base()


class MyCallback(Base):
    __tablename__ = 't_callback'
    c_date = Column(String(8), primary_key=True)
    c_time = Column(String(8), primary_key=True)
    c_ani = Column(String, primary_key=True)
    c_dnis = Column(String, nullable=False)
    c_callback_num = Column(String)
    c_callback_date = Column(String(8))
    c_callback_time = Column(String(8))
    c_complete = Column(String(1),index=True)  #Y:completed , N:Not completed
    c_del = Column(String(1))       #Y:deleted , N:Not deleted

#callback_index = Index('callback_idx', callback.c_complete)

Base.metadata.create_all(engine) 

<create_db.py>

If you run the above python code, perhaps you can see that the table is created like this:

root@ubuntusrv:/usr/local/src/study/flask_sqlite# sqlite3 callback.db 
SQLite version 3.31.1 2020-01-27 19:55:54
Enter ".help" for usage hints.
sqlite> select * from sqlite_master;
table|t_callback|t_callback|2|CREATE TABLE t_callback (
	c_date VARCHAR(8) NOT NULL, 
	c_time VARCHAR(8) NOT NULL, 
	c_ani VARCHAR NOT NULL, 
	c_dnis VARCHAR NOT NULL, 
	c_callback_num VARCHAR, 
	c_callback_date VARCHAR(8), 
	c_callback_time VARCHAR(8), 
	c_complete VARCHAR(1), 
	c_del VARCHAR(1), 
	PRIMARY KEY (c_date, c_time, c_ani)
)
index|sqlite_autoindex_t_callback_1|t_callback|3|
index|ix_t_callback_c_complete|t_callback|4|CREATE INDEX ix_t_callback_c_complete ON t_callback (c_complete)
sqlite>

 Be Careful : The primary key must exist in the table to use SQLAlchemy.


The following is app.py using Flask. In the previous MySQL example, we used several Python codes, but this time we will only use one file.

# app.py
from flask import Flask, render_template, request, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////usr/local/src/study/flask_sqlite/callback.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = "flask rocks!"

db = SQLAlchemy(app)

#Creating model table for our CRUD database
class MyCallback(db.Model):
    __tablename__ = 't_callback'
    c_date = db.Column(db.String(8), primary_key=True)
    c_time = db.Column(db.String(8), primary_key=True)
    c_ani = db.Column(db.String, primary_key=True)
    c_dnis = db.Column(db.String, nullable=False)
    c_callback_num = db.Column(db.String)
    c_callback_date = db.Column(db.String(8))
    c_callback_time = db.Column(db.String(8))
    c_complete = db.Column(db.String(1),index=True)  #Y:completed , N:Not completed
    c_del = db.Column(db.String(1))       #Y:deleted , N:Not deleted
    def __init__(self, date, ani, dnis, callback_num):
       self.c_date = date
       self.c_time = time
       self.c_ani = ani
       self.c_dnis = dnis
       self.c_callback_num = callback_num
       self.c_callback_date = ''
       self.c_complete = 'N'
       self.c_del = 'N'

#This is the index route where we are going to
#query on all our employee data
@app.route('/')
def Index():
    all_data = MyCallback.query.all()
 
    return render_template("index.html", t_callback = all_data)
    # return render_template("test.html")


#this is our update route where we are going to update our employee
@app.route('/update/<c_date>/<c_time>/<c_ani>/', methods = ['GET', 'POST'])
def update(c_date, c_time, c_ani):
    now = datetime.datetime.now()
    formattedDate = now.strftime("%Y%m%d")    
    formattedTime = now.strftime("%H%M%S")    
    qry = db.session.query(MyCallback).filter(MyCallback.c_date==c_date).filter(MyCallback.c_time==c_time).filter(MyCallback.c_ani==c_ani)
    my_data = qry.first()
    my_data.c_complete = 'Y'
    my_data.c_callback_date =  formattedDate
    my_data.c_callback_time =  formattedTime
    db.session.commit()
    return redirect(url_for('Index'))

 
 
 
#This route is for deleting our employee
@app.route('/delete/<c_date>/<c_time>/<c_ani>/', methods = ['GET', 'POST'])
def delete(c_date, c_time, c_ani):
    qry = db.session.query(MyCallback).filter(MyCallback.c_date==c_date).filter(MyCallback.c_time==c_time).filter(MyCallback.c_ani==c_ani)
    my_data = qry.first()
    db.session.delete(my_data)
    db.session.commit()
    flash("Delete callback success")
    print("Delete callback success:" + c_date + ' ' + c_time + ' ' + c_ani)
    return redirect(url_for('Index'))

if __name__ == "__main__":
    app.run(host="0.0.0.0", debug=True) 


<app.py>


These are html template files.

<!DOCTYPE html>
<html lang="en">
<head>
    <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.4.1/css/bootstrap.min.css" integrity="sha384-Vkoo8x4CGsO3+Hhxv8T/Q5PaXtkKtu6ug5TOeNV6gBiFeWPGFN9MuhOf23Q9Ifjh" crossorigin="anonymous">
    <meta charset="UTF-8">
    <title>{% block title %} {% endblock %} </title>
</head>
<body>

{% block body %} {% endblock %}

<script src="https://code.jquery.com/jquery-3.4.1.slim.min.js" integrity="sha384-J6qa4849blE2+poT4WnyKhv5vZF5SrPo0iEjwBvKU7imGFAV0wwj1yYfoRSJoZ+n" crossorigin="anonymous"></script>
<script src="https://cdn.jsdelivr.net/npm/popper.js@1.16.0/dist/umd/popper.min.js" integrity="sha384-Q6E9RHvbIyZFJoft+2mJbHaEWldlvI9IOYy5n3zV9zzTtmI3UksdQRVvoxMfooAo" crossorigin="anonymous"></script>
<script src="https://stackpath.bootstrapcdn.com/bootstrap/4.4.1/js/bootstrap.min.js" integrity="sha384-wfSDF2E50Y2D1uUdj0O3uMBJnjuUD4Ih7YwaYd1iqfktj0Uod8GCExl3Og8ifwB6" crossorigin="anonymous"></script>

</body>
</html>

<base.html>


{% extends 'base.html' %}

{% block title %} Callback Request  {% endblock %}

{% block body %}

<div class="jumbotron p-3">
    <div class="well text-center">
        <h1>Callback List</h1>
    </div>

</div>

{% endblock %}

<header.html>


{% extends 'base.html' %}
{% include 'header.html' %}

{% block title %} Home {% endblock %}

{% block body %}

<div class="container">
    <div class="row">
        <div class="col md-12">

            <div class="jumbotron p-10">

                {% with messages = get_flashed_messages() %}
                {% if messages %}
                {% for message in messages %}
                {% endfor %}
                {% endif %}
                {% endwith %}

                <table class="table table-hover table-dark">

                    <tr>

                        <th>DATE</th>
                        <th>TIME</th>
                        <th>ANI</th>
                        <th>DNIS</th>
                         <th>Callback Phone</th>
                        <th>Callback Date</th>
                        <th>Callback Time</th>
                        <th>Done</th>

                    </tr>


                   {% for row in t_callback %}
                    <tr>
                        <td>{{row.c_date}}</td>
                        <td>{{row.c_time}}</td>
                        <td>{{row.c_ani}}</td>
                        <td>{{row.c_dnis}}</td>
                        <td>{{row.c_callback_num}}</td>
                        <td>{{row.c_callback_date}}</td>
                        <td>{{row.c_callback_time}}</td>
                        <td>{{row.c_complete}}</td>

                        <td>
                            <!-- <a href="/update/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-warning btn-xs" onclick="return confirm('콜백처리를 완료하시겠습니까 ?')">처리 완료</a>
                            <a href="/delete/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-danger btn-xs" onclick="return confirm('콜백 요청 기록을 삭제하시겠습니까 ?')">삭제</a> -->
                            <a href="/update/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-warning btn-xs" onclick="return confirm('Do you want to complete callback processing?')">complete</a>
                            <a href="/delete/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-danger btn-xs" onclick="return confirm('Are you sure you want to delete the callback request record ?')"> delete </a>
                        </td>
                    </tr>

                    {% endfor %}
                </table>
            </div>

        </div>
    </div>
</div>


{% endblock %}

<index.html>

The directory structure is as follows.

root@ubuntusrv:/usr/local/src/study/flask_sqlite# tree
.
├── app.py
├── callback.db
├── create_db.py
└── templates
    ├── base.html
    ├── header.html
    └── index.html

1 directories, 11 files


For testing, I put the data in advance using the SQLite Expert tool.

<SQLite Expert GUI Tool>


Now run the flask app and check the result.

root@ubuntusrv:/usr/local/src/study/flask_sqlite# python3 app.py 
 * Serving Flask app "app" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 717-309-215

You can see that the Flask application ran normally and can be accessed through port 5000. I will try to access it using a browser on a PC.

<Web Browser screen>


Add datepicker

Finally, let's add a part to the callback list that queries using date conditions. 


# app.py
from flask import Flask, render_template, request, redirect, url_for, flash, session
from flask_sqlalchemy import SQLAlchemy
import datetime

from flask_wtf import FlaskForm
from wtforms.fields.html5 import DateField
from wtforms.validators import DataRequired
from wtforms import validators, SubmitField

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////usr/local/src/study/flask_sqlite/callback.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = "flask rocks!"

db = SQLAlchemy(app)

#DatePicker
class InfoForm(FlaskForm):
    startdate = DateField('Start Date', format='%Y-%m-%d', validators=(validators.DataRequired(),))
    enddate = DateField('End Date', format='%Y-%m-%d', validators=(validators.DataRequired(),))
    submit = SubmitField('Submit')

#Creating model table for our CRUD database
class MyCallback(db.Model):
    __tablename__ = 't_callback'
    c_date = db.Column(db.String(8), primary_key=True)
    c_time = db.Column(db.String(8), primary_key=True)
    c_ani = db.Column(db.String, primary_key=True)
    c_dnis = db.Column(db.String, nullable=False)
    c_callback_num = db.Column(db.String)
    c_callback_date = db.Column(db.String(8))
    c_callback_time = db.Column(db.String(8))
    c_complete = db.Column(db.String(1),index=True)  #Y:completed , N:Not completed
    c_del = db.Column(db.String(1))       #Y:deleted , N:Not deleted
    def __init__(self, date, ani, dnis, callback_num):
       self.c_date = date
       self.c_time = time
       self.c_ani = ani
       self.c_dnis = dnis
       self.c_callback_num = callback_num
       self.c_callback_date = ''
       self.c_complete = 'N'
       self.c_del = 'N'

#This is the index route where we are going to
#query on all our employee data
@app.route('/', methods=['GET','POST'])
def Index():
    form = InfoForm()
    if form.validate_on_submit():
        session['startdate'] = str(form.startdate.data)[:4] +  str(form.startdate.data)[5:7] + str(form.startdate.data)[-2:]      #2021-02-02
        session['enddate'] = str(form.enddate.data)[:4] +  str(form.enddate.data)[5:7] + str(form.enddate.data)[-2:]
        print(session['startdate'])
        print(session['enddate'])
        all_data = db.session.query(MyCallback).filter(MyCallback.c_date <= session['enddate']).filter(MyCallback.c_date >= session['startdate'])
    else:    
        all_data = MyCallback.query.all()
    print('Query Count:%d'%all_data.count())
    return render_template("index.html", t_callback = all_data, form=form, count = all_data.count() )
    # return render_template("test.html")


#this is our update route where we are going to update our employee
@app.route('/update/<c_date>/<c_time>/<c_ani>/', methods = ['GET', 'POST'])
def update(c_date, c_time, c_ani):
    now = datetime.datetime.now()
    formattedDate = now.strftime("%Y%m%d")    
    formattedTime = now.strftime("%H%M%S")    
    qry = db.session.query(MyCallback).filter(MyCallback.c_date==c_date).filter(MyCallback.c_time==c_time).filter(MyCallback.c_ani==c_ani)
    my_data = qry.first()
    my_data.c_complete = 'Y'
    my_data.c_callback_date =  formattedDate
    my_data.c_callback_time =  formattedTime
    db.session.commit()
    return redirect(url_for('Index'))

 
 
 
#This route is for deleting our employee
@app.route('/delete/<c_date>/<c_time>/<c_ani>/', methods = ['GET', 'POST'])
def delete(c_date, c_time, c_ani):
    qry = db.session.query(MyCallback).filter(MyCallback.c_date==c_date).filter(MyCallback.c_time==c_time).filter(MyCallback.c_ani==c_ani)
    my_data = qry.first()
    db.session.delete(my_data)
    db.session.commit()
    flash("Delet Callback success")
    print("Delete Callback success :" + c_date + ' ' + c_time + ' ' + c_ani)
    return redirect(url_for('Index'))

if __name__ == "__main__":
    app.run(host="0.0.0.0", debug=True)    

<app.py>



{% extends 'base.html' %}
{% include 'header.html' %}

{% block title %} Home {% endblock %}

{% block body %}


<div class="container">
    <div class="row">
        <div class="col md-12">

            <div class="jumbotron p-10">
                <!-- <form action="" method='POST' name="myform">
                    {{ form.csrt_token }}
                    {{ form.startdate.label }} {{form.startdate(class='datepicker')}} {{form.hidden_tag()}}
                    {{ form.enddate.label }} {{form.enddate(class='datepicker')}} {{form.hidden_tag()}}
                    {{form.submit()}}
                </form> -->
                <form action="" method='POST' name="myform">
                    {{ form.csrt_token }}
                    {{ form.startdate.label }}  &nbsp;&nbsp; {{form.startdate(class='datepicker')}} &nbsp;&nbsp;&nbsp; {{form.hidden_tag()}}
                    {{ form.enddate.label }} &nbsp;&nbsp;{{form.enddate(class='datepicker')}} &nbsp;&nbsp;&nbsp;{{form.hidden_tag()}} &nbsp;&nbsp;&nbsp;
                    <input class="btn btn-primary" type="submit" value="submit" >
                </form>

                <br>    

                {% with messages = get_flashed_messages() %}
                {% if messages %}
                {% for message in messages %}
                {% endfor %}
                {% endif %}
                {% endwith %}

                <table class="table table-hover table-dark">

                    <tr>

                        <th>DATE</th>
                        <th>TIME</th>
                        <th>ANI</th>
                        <th>DNIS</th>
                        <!-- <th>콜백요청번호</th>
                        <th>콜백일자</th>
                        <th>콜백시간</th> -->
                        <th>Callback Phone</th>
                        <th>Callback Date</th>
                        <th>Callback Time</th>
                        <th>Done</th>

                    </tr>


                   {% for row in t_callback %}
                    <tr>
                        <td>{{row.c_date}}</td>
                        <td>{{row.c_time}}</td>
                        <td>{{row.c_ani}}</td>
                        <td>{{row.c_dnis}}</td>
                        <td>{{row.c_callback_num}}</td>
                        <td>{{row.c_callback_date}}</td>
                        <td>{{row.c_callback_time}}</td>
                        <td>{{row.c_complete}}</td>

                        <td>
                            <!-- <a href="/update/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-warning btn-xs" onclick="return confirm('콜백처리를 완료하시겠습니까 ?')">처리 완료</a>
                            <a href="/delete/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-danger btn-xs" onclick="return confirm('콜백 요청 기록을 삭제하시겠습니까 ?')">삭제</a> -->
                            <a href="/update/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-warning btn-xs" onclick="return confirm('Do you want to complete callback processing?')">complete</a>
                            <a href="/delete/{{row.c_date}}/{{row.c_time}}/{{row.c_ani}}" class="btn btn-danger btn-xs" onclick="return confirm('Are you sure you want to delete the callback request record ?')"> delete </a>
                        </td>
                    </tr>

                    {% endfor %}
                </table>
                <h4 class="text-primary">Total {{count}} rows searched.</h4>

            </div>

        </div>
    </div>
</div>

{% endblock %}

<index.html>

As shown in the following figure, a DatePicker that can be used for query conditions is well made.

<Add datepicker to filter the callback rows>


Wrapping up

You can download the source code from my Github. You can see that the complete and delete buttons also work normally. In Parwiz's article, there is a content that works even the modal dialog. For more details, please refer to Parwiz's article..


















댓글

이 블로그의 인기 게시물

MQTT - C/C++ Client

RabbitMQ - C++ Client #1 : Installing C/C++ Libraries

C/C++ - Everything about time, date