SQLite Skills
SQLite is covered in more detail here.
Note: When you connect to an SQLite database file that does not exist, SQLite creates a new database for you.
sqlite_a.py
from flask import Flask
import sqlite3
app = Flask(__name__)
###------------- sql statements:
sql_create = """
CREATE TABLE IF NOT EXISTS orders (
num INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
flavour TEXT,
quantity INTEGER
);
"""
sql_insert = """INSERT OR IGNORE INTO orders (
flavour, quantity) values (?,?)""" #num not needed - AI
sql_select = "SELECT * FROM orders"
###------------- on server initialise:
db = sqlite3.connect('orders.db') #create or connect to db
db.cursor().execute(sql_create) #create table if not exists
db.cursor().execute(sql_insert, ("ham & cheese", 5)) #insert tuple
try:
db.commit() #save
except Exception:
db.rollback() #crash
finally:
db.close()
###------------- on application landing:
@app.route("/")
def main():
html = ""
try:
db = sqlite3.connect('orders.db')
html = str(db.cursor().execute(sql_select).fetchone())
except Exception as error_msg:
html = str(error_msg)
finally:
db.close()
return html
app.run(debug=True)
sqlite_b.py
from flask import Flask
import sqlite3
app = Flask(__name__)
###------------- data:
orders = [
("ham & cheese", 5),
("cheese & bacon", 3),
("vegetarian", 7),
]
###------------- sql statements:
sql_create = """
CREATE TABLE IF NOT EXISTS orders (
num INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
flavour TEXT,
quantity INTEGER
);
"""
sql_insert = """INSERT INTO orders (
flavour, quantity) values (?,?)"""
#num not needed as auto increment.
#Note this also won't violate any integrity constraint,
#because each record is assigned a new num (as AI),
#so 'OR IGNORE' (after INSERT) is redundant here.
sql_select = "SELECT * FROM orders"
###------------- on server initialise:
db = sqlite3.connect('orders.db') #create or connect to db
db.cursor().execute(sql_create) #create table if not exists
db.cursor().executemany(sql_insert, orders) #insert tuple
try:
db.commit() #save
except Exception:
db.rollback() #crash
finally:
db.close()
###------------- on application landing:
@app.route("/")
def main():
html = ""
try:
db = sqlite3.connect('orders.db')
html = str(db.cursor().execute(sql_select).fetchall())
except Exception as error_msg:
html = str(error_msg)
finally:
db.close()
return html
app.run()
#app.run(debug=True)
#^^ You can run in debug mode,
#however each time the server restarts the records will duplicate.
sqlite_c.py
from flask import Flask
import sqlite3
app = Flask(__name__)
###------------- data:
orders = [
("ham & cheese", 5),
("cheese & bacon", 3),
("vegetarian", 7)
]
###------------- sql:
sql_create = """
CREATE TABLE IF NOT EXISTS orders (
num INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
flavour TEXT,
quantity INTEGER
);
"""
sql_insert = """INSERT INTO orders (
flavour, quantity) values (?,?)"""
sql_select = "SELECT * FROM orders"
###------------- server init:
db = sqlite3.connect('orders.db')
db.cursor().execute(sql_create)
db.cursor().executemany(sql_insert, orders)
try:
db.commit()
except Exception:
db.rollback()
finally:
db.close()
###------------- landing:
@app.route("/")
def main():
html = ""
try:
db = sqlite3.connect('orders.db')
db.row_factory = sqlite3.Row #access tuples by column name
records = db.cursor().execute(sql_select).fetchall()
for row in records:
html += row["flavour"] + "<br>"
except Exception as error_msg:
html = str(error_msg)
finally:
db.close()
return html
app.run()
#app.run(debug=True)
#^^ You can run in debug mode,
#however each time the server restarts the records will duplicate.
sqlite_d.py
from flask import Flask, render_template_string
import sqlite3
app = Flask(__name__)
###------------- data:
orders = [
("ham & cheese", 5),
("cheese & bacon", 3),
("vegetarian", 7)
]
###------------- sql:
sql_create = """
CREATE TABLE IF NOT EXISTS orders (
num INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
flavour TEXT,
quantity INTEGER
);
"""
sql_insert = """INSERT INTO orders (
flavour, quantity) values (?,?)"""
sql_select = "SELECT * FROM orders"
###------------- server init:
db = sqlite3.connect('orders.db')
db.cursor().execute(sql_create)
db.cursor().executemany(sql_insert, orders)
try:
db.commit()
except Exception:
db.rollback()
finally:
db.close()
###------------- landing template:
html = """
{% for each in orders %}
{{each[0]}}, flav: {{each[1]}}, qty: {{each[2]}}<br>
{% endfor %}
"""
###------------- landing:
@app.route("/")
def main():
try:
db = sqlite3.connect('orders.db')
records = db.cursor().execute(sql_select).fetchall()
except Exception as error_msg:
return str(error_msg) # ... better way to handle this
finally:
db.close()
return render_template_string(html, orders = records)
app.run()
#app.run(debug=True)
#^^ You can run in debug mode,
#however each time the server restarts the records will duplicate.
sqlite_e.py
from flask import Flask, render_template_string
import sqlite3
app = Flask(__name__)
###------------- data:
orders = [
("ham & cheese", 5),
("cheese & bacon", 3),
("vegetarian", 7)
]
###------------- sql:
sql_create = """
CREATE TABLE IF NOT EXISTS orders (
num INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
flavour TEXT,
quantity INTEGER
);
"""
sql_insert = """INSERT INTO orders (
flavour, quantity) values (?,?)"""
sql_select = "SELECT * FROM orders"
###------------- server init:
db = sqlite3.connect('orders.db')
db.cursor().execute(sql_create)
db.cursor().executemany(sql_insert, orders)
try:
db.commit()
except Exception:
db.rollback()
finally:
db.close()
###------------- landing template:
html = """
<h1>Records by index</h2>
{% for each in rec_by_index %}
{{each[0]}}, flav: {{each[1]}}, qty: {{each[2]}}<br>
{% endfor %}
<br>
<h1>Records by field</h2>
{% for each in rec_by_field %}
{{each["num"]}}, flav: {{each["flavour"]}}, qty: {{each["quantity"]}}<br>
{% endfor %}
"""
###------------- landing:
@app.route("/")
def main():
try:
db = sqlite3.connect('orders.db')
records_by_index = db.cursor().execute(sql_select).fetchall()
#Now access tuples by column name:
#---------------------------------
db.row_factory = sqlite3.Row
#---------------------------------
records_by_field = db.cursor().execute(sql_select).fetchall()
except Exception as error_msg:
return str(error_msg) # ... better way to handle this
finally:
db.close()
return render_template_string(html,
rec_by_index = records_by_index,
rec_by_field = records_by_field)
app.run()
#app.run(debug=True)
#^^ You can run in debug mode,
#however each time the server restarts the records will duplicate.