import html
import os
import sys
from flask import Flask
from flask import redirect
from flask import request
from flask import url_for
from flask_sqlalchemy import SQLAlchemy
from jinja2 import Environment
from sqlalchemy import inspect
if 'AGAMA_DATABASE_URI' not in os.environ:
print('''
ERROR: Environment variable AGAMA_DATABASE_URI is not set.
AGAMA_DATABASE_URI uses the SQLAlchemy format.
For SQLite3 use
AGAMA_DATABASE_URI=sqlite:////path/to/db.sqlite3 # yes, 4 slashes
For MySQL use
AGAMA_DATABASE_URI=mysql://:@/
For other examples see
https://docs.sqlalchemy.org/en/13/core/engines.html#database-urls.''')
sys.exit(1)
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ['AGAMA_DATABASE_URI']
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class Item(db.Model):
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
value = db.Column(db.String(255), unique=True, nullable=False)
state = db.Column(db.Integer, default=0)
@app.before_request
def before_request():
try:
init_db()
except Exception:
db.session.remove()
db.engine.dispose()
init_db()
@app.route('/')
def index():
return html_render(items=Item.query.all())
@app.route('/items/add', methods=['GET', 'POST'])
def item_add():
item = request.form['new_item']
if len(item) > 255:
return html_error('The item you are trying to add seems too large; it should be shorter than 256 characters.')
if Item.query.count() >= 100:
return html_error('You are trying to add too many items; you have 100 items added already.')
existing_item = Item.query.filter_by(value=item).first()
if existing_item:
return html_error('Item [%s] already exists.' % item)
if request.method == 'POST':
app.logger.info("Adding item '%s'..." % item)
db.session.add(Item(value=item))
db.session.commit()
return redirect(url_for('index'))
@app.route('/items//delete')
def item_delete(id):
item = Item.query.filter_by(id=id)
if item:
app.logger.info('Deleting item %s...' % id)
item.delete()
db.session.commit()
return redirect(url_for('index'))
@app.route('/items//swap-state')
def item_swap_state(id):
item = Item.query.get(id)
if item:
app.logger.info('Swapping item %s state (current state: %s)...' % (id, item.state))
item.state = 0 if item.state else 1
db.session.commit()
return redirect(url_for('index'))
def html_error(error_msg):
error_msg_html = '
""").render(host=os.uname()[1], items=items)
def init_db():
with db.engine.connect() as conn:
if not inspect(conn).has_table(Item.__tablename__):
app.logger.info('Initializing the database...')
db.create_all()
db.session.add(Item(value="I'm a demo item — click me to change my state", state=1))
db.session.add(Item(value="I'm another demo item — click ❌ to remove me →"))
db.session.commit()
if __name__ == '__main__':
app.run()