Tag Archives: python flask

Python Flask activity logging

This post describes a simple pproach to recording login, logout, insert, update and delete operations that are carried out by a Python Flask application.

This is not the same as a table auditing function similar to Ruby’s Paperclip module.

https://github.com/slugbucket/crossword-hints/issues/9 is an example feature branch for a personal project implementing this.

The activity log will be stored in a database table with the following structure:
* rowid – auto-assigned unique id for the activity record
* actor – the name of the user performing the action
* action – one of login, insert, update, delete or logout,
* item_type -the table on which the operation has been performed.
* item_id – the numeric id of the item under operation
* act_action – details of the content that has been changed

This can be modeled in Flask using PeeWee

...
from peewee import *
from datetime import date, timedelta, datetime

...

class BaseModel(Model):
    with application.app_context():
    class Meta:
        database = database

    class activity_logs(BaseModel):
        rowid = AutoField()
        actor = CharField(max_length=32)
        action = CharField(max_length=32)
        item_type = CharField(max_length=32)
        item_id = IntegerField()
        act_action = TextField()
        created_at = CharField(max_length=32)
        updated_at = DateTimeField(default=datetime.now())

The application context is called application to work better when deployed to AWS.

Having used Flask-login to control the login process. we have a variable. current_user, that contains a numeric id of the user from their record in teh database.

The users model includes a method to return the name of the user and this can be used for the ‘actor’ value.

A simple function can create the activity log reord:

def add_log(actor, action, item_type, item_id, activity):
    log = activity_logs(actor=actor,
          action=action,
          item_type=item_type,
          item_id=item_id,
          act_action=activity,
          created_at=datetime.now(),
          updated_at=datetime.now())
    log.save()

And activities to be logged can be logged with something like,

log = ("name: %s\ndescription: %s" % (name, description))
add_log(users.get_name(current_user), 'update', 'solution_types', id, log)

Where,

  • name and description are variables containing form-submitted data
  • activity_logs is the PeeWee model of the databse table
  • users is a PeeWee ORM model of an application user that can be authenticated.

When used the (sqlite3) database records will look like,

sqlite> select * from activity_logs;
crossy|login|user|1|Successful login for crossy|2019-02-15 18:44:38.257050|2019-02-15 18:44:38.257061
crossy|update|solution_types|12|name: Homophone
description: Sounds like a word that has a different spelling|2019-02-15 18:45:03.227188|2019-02-15 18:45:03.227193
crossy|logout|user|1|Successful logout user for crossy|2019-02-15 18:48:11.043243|2019-02-15 18:48:11.043249

With these records it should be possible to display an activity report and perhaps even recover previous versions of particular database records.

Improvements

  • Storing the full record on each update is inefficient; only differences should be recorded.
  • The logging function should be able to determine what to include in the log message based on the

The new submission form

This post will detail the processing of a new submission form that contains a multiple select form element.

A widget is described as a set of versioned components where a component version could be used in many widgets:
components:
id: INT
name: STRING
description: TEXT

component_versions:
id: INT
name: STRING
component_id: INT

widgets:
id: INT
name: STRING
description: TEXT

component_version_widgets:
component_id: IN
widget_id: INT

A simple route to handle a request to create  new item could be something like:

@app.route('/widgets/new', methods=['GET', 'POST'])
def newWidget():
    # Wasn't able to find any real documentation or examples of how to use the concat
    # function but sort of figured it out by looking at the PeeWee source code. This
    # uses a local function to create a list of list of lists of the items to be
    # displayed in the multiple select element in the form
    # [ [1, <component_name>-<component_version> ], [2, <component_name>-<component_version> ], ... ]
    # that is hopefully easier to process in the template 
    cvids = dict2lol(component_versions
            .select(component_versions.id.alias("cmpntid"),
             components.name.concat("-").concat(component_versions.name).alias("cmpntname"))
             .join(components, JOIN.INNER,
             on=component_versions.component_id == components.id)
             .dicts(), 'cmpntid', 'cmpntname')
    if request.method == 'POST':
        (status, msg) = validateWidget(request.form)
         # Use .getlist() to retrieve the multiple select items
         slctd = request.form.getlist('component_version_ids')
         if status:
             # Although auto-increment works with SQLite3 it didn't seem to work
             # when creating records with PeeWee
             nextId = widgets.select(fn.Max(widgets.id)+1).scalar()
             widgets.create(id=nextId,
                                         name=request.form['name'], \
                                         description=request.form['description'], \
                                         updated_by="config.admin", \
                                          updated_at=datetime.datetime.now())
             for cvid in slctd:
                  component_version_widget.create(widget_id=nextId, component_version_id=cvid)
                  logActivity('create', 'component_version_widgets', nextId, ("component_version_id: %s" % pvid))

             logActivity('create', 'widgets', nextId, request.form['name'])
              flash(("Saved new widget, update for %s." % request.form['name']))
              return redirect('/widgets')
        else:
             flash("Widget creation failed: %s." % msg)
              slctd = []
              for pv in request.form.getlist('component_version_ids'):
                 slctd.append(int(pv))
                 wdgt = {"description": request.form['description'],
                           "name": request.form['name'],
                           "component_version_ids": slctd}
    else:
        wdgt = {'name': 'Widget name', 'description': 'Brief description', "component_version_ids": []}
    return render_template('views/widgets/new.html', wdgt=wdgt, cvids=cvids, req=request)

Notes:

  • cvids: a list of lists of select options in the form [id, name]. Possibly being
    lazy but this is a bit simpler than trying to process what might come out of a PeeWee
    result set
  • slctd: a list of selected options from the select form; either from the database or the
    submitted form data. When creating this list from the submitted form we need to cast the id values as int’s or they won’t be recognised by the template inline conditional.
  • wdgt: dict containing the values for the template form items
  • formatting python code in the edit window is quite tricky so no guarantees that it compiles cleanly

The template to display this might look a bit like,

<form name="widgets" method="POST" action="{{ req.path }}">

<p>
Widget Widget name Description {{ rel['description'] }}
Component versions:
{%- for opt in cvids-%} {{ opt[1] }} {%- endfor -%}
</p>
<input type="submit" name="submit" value="Save widget"> </form>

The only real point to note in here is the method by which the ‘selected’ attribute is added to the option in the select list: the inline if.

References

The edit form submission

This post will detail the processing of an edit submission form that contains a multiple select form element.

A widget is described as a set of versioned components where a component version could be used in many widgets:

components:
id: INT
name: STRING
description: TEXT

component_versions:
id: INT
name: STRING
component_id: INT

widgets:
id: INT
name: STRING
description: TEXT

component_version_widgets:
component_id: INT
widget_id: INT

@app.route('/widgets/<int:id>/edit', methods=['GET', 'POST'])
def editWidget(id):
    # Get the component names and versions and format them for easy display
    # in the template
    cvids = dict2lol(component_versions
                .select(component_versions.id.alias("cmpntverid"),
                 components.name.concat("-").concat(component_versions.name).alias("cmpntname"))
                 .join(components, JOIN.INNER,
                 on=component_versions.component_id == components.id)
                 .dicts(), 'cmpntverid', 'cmpntname')

    if request.method == 'POST':
         (status, msg) = validateWidget(request.form)
         slctd = request.form.getlist('component_version_ids')
         if status:
             savewdgt = widgets(id=id,
                       name=request.form['name'], \
                       description=request.form['description'], \
                       updated_by="config.admin", \
                       updated_at=datetime.datetime.now())
             savewdgt.save()
            # Delete the current component versions for this widget and save
             # the values submitted in the form
             qry = component_version_widgets.delete().where(component_version_widgets.widget_id == id)
             qry.execute()
             for cvid in slctd:
                 component_version_widgets.create(widget_id=id, component_version_id=cvid)
 logActivity('update', 'component_version_widgets', id, ("component_version_id: %s" % cvid))

                logActivity('update', 'widgets', id, request.form['name'])
            flash(("Saved update for widget, %s." % request.form['name']))
            return redirect('/widgets')
        else:
             flash(("widget update failed: %s." % msg))
             slctd = []
             for cv in request.form.getlist('component_version_ids'):
                slctd.append(int(cv))
                wdgt = {"description": request.form['description'], "name": request.form['name'],
 "component_version_ids": slctd}
    else:
        # Display the initial edit form with details from the database
         # Get the details of the item to be edited 
         try:
             rs = widgets.select().where(widgets.id == id).get()
         except DoesNotExist:
             flash(("Cannot locate widget record with id = %s" % id))
         return redirect('/widgets')


    slctd = []
    for cvid in component_version_widgets \
           .select(component_version_widgets.component_version_id) \
           .where(component_version_widgets.widget_id == id):
        slctd.append(cvid.component_version_id)
        wdgt = {'name': rs.name, 'description': rs.description, 'component_version_ids': slctd}

    return render_template('views/widgets/edit.html', wdgt=wdgt, cvids=cvids, req=request)

Notes:

  • cvids: a list of lists of select options in the form [id, name]. Possibly being lazy but this is a bit simpler than trying to process what might come out of a PeeWee result set
  • slctd: a list of selected options from the select form; either from the database or the
    submitted form data. When creating this list from the submitted form we need to cast the id values as int’s or they won’t be recognised by the template inline conditional.
  • wdgt: dict containing the values for the template form items

The dict2lol function is simply:

def dict2lol(rs, val, text):
    lol = []
    for row in rs:
        lol.append([row[val], row[text]])
    return(lol)

The template to display this might look a bit like,

<form name="widgets" method="POST" action="{{ req.path }}">
<p>
Widget Widget name Description {{ rel['description'] }}
Component versions:
{%- for opt in cvids-%} {{ opt[1] }} {%- endfor -%}
</p>
<input type="submit" name="submit" value="Save widget"> </form>

The only real point to note in here is the method by which the ‘selected’
attribute is added to the option in the select list: the inline if.

References

POST form handling with Python Flask

A series of posts describing how to handle POST form submission with Python Flask.

Flask is a great framework for building simple HTTP-based applications but without the overhead and baggage of something like Django; Sinatra provides a similar situation when compared to Ruby on Rails.

One of the great things about Rails is how easy it is to work with many-to-many relationships and multiple select form elements.

But having abandoned working with Rails (and with no urge to go back) but managing to use some of the structure in Python Flask apps I do have to devise a method for handling new and edit routes using PeeWee as an ORM.

Most of this is probably self-evident but figuring out how to work with the multiple select form elements was quite tricky.

Simple POST form processing

The general form for a simple (single form) new page could be routes that accept GET and POST requests:

@app.route('/widgets/new', methods=['GET', 'POST'])
def newWidget():
  • If the request.method is ‘POST’ this is a form submission:
    • validate the form data and return a tuple containing the status and any error message
    • If the status id good:
      • Create a new record in the database
      • If processing a multiple select element:
      • for each item in the multiple select create a new many-to-many table entry
      • Record the activity in the log table
    • generate a flash message for display on the next page
    • redirect to the index page
    • If the status is bad:
      • Create a flash message with the error from the validation check
      • create a dict containing the submitted form elements to be used in the template
  • If not a POST request
    • prepare a dict containing default values for the form template

In all cases:

  • Generate objects to include database content to be passed to the temaplate
  • render the template

The general form for a simple (single form) edit page could be

  • Retrieve the edited object’s details from the database
  • If the request.method is ‘POST’ this is a form submission:
    • validate the form data and return a tuple containing the status and any error message
    • If the status id good:
      • Create a new record in the database
      • Record the activity in the log table
      • If processing a multiple select element:
        • delete the current many-to-many records the match the id of the edited object
        • for each item in the multiple select create a new many-to-many table entry
        • Record the activity in the log table
      • generate a flash message for display on the next page
      • redirect to the index page
    • If the status is bad:
      • Create a flash message with the error from the validation check
      • create a dict containing the submitted form elements to be used in the template
  • If not a POST request
    • prepare a dict containing edited item’s values for the form template

In all cases:

  • Generate objects to include database content to be passed to the template
  • render the template

    So far, so good, and I’m sure there’s nothing revelatory in any of this. The following posts will include some simple example code to demonstrate these types of form processing.

SQLite3 autoincrement primary and foreign keys

Just a wee nugget I picked up from the PeeWee documentation that needs to be saved for later reference ‘cos I’m bound to forget this with later projects and I’m including unnecessary code in my applications.

I had cursed PeeWee for not appearing to support incrementing primary keys when inserting new records and having to include a line to calculate the next insert id via a query (not a safe operation, I know), but it turns out that there is an alternate way to specify a primary key in the PeeWee model:

id = PrimryKeyField()

rather than

id = IntegerField(primary_key=True)

(though like most ORMs it doesn’t think you need the id column; I prefer to see it). This comes, however, from a Playhouse extension which I want to avoid.

It also seems that foreign keys are not honored by default and that they have to be explicity requested with,

database = SqliteDatabase(app.config['DATABASE'], pragmas=((‘foreign_keys’, ‘on’),))

when making the database connection. It doesn’t break my run of unit tests which is also a bonus.

Finally, a compound primary is better specified in the model as,

class Meta:
    primary_key = CompositeKey('column1', 'column2')

rather than  unique index for the unit tests to still work; the column names don’t include the ‘_id’ part of what might actually be in the database.

References

Python Flask, PeeWee and unittest

With all the web projects I work on I aim to have learned something new by the end (there’s evidence that once I have learned ‘that thing’ I rapidly lose interest in the actual project and move onto something else, but I am try to be better): with my latest (work-related site, it was to have another go at developing a test suite with  the webapp.

With this being an internal work project, I’m not at liberty to share the details, but I did find a way of developing a set of unit tests to run against a Flask application that uses the PeeWee  ORM; my previous attempt ended up with copying the whole application file and trying to modify it so that tests could be run. Not a winning strategy.

I’m not going to in any way criticise the marvellous work that goes into the Flask ecosphere and I really enjoy developing applications with it but the docs can be a littke frustrating to work with if you’re not already an expert (and why SO is so important for specific tasks) because there are many ways to skin the problem; finding the correct method that’s common to all 3 libraries is a real challenge, and the docs do actually hint at this. And why it’s worth a blog post so that I have something to refer back to in future.

The typical startup code for the Flask application might look something like,

from flask import Flask, request, flash, redirect, render_template, g, jsonify, Response, send_file
from werkzeug import Headers
from jinja2 import Environment, PackageLoader, select_autoescape
import sqlite3

app = Flask(__name__)
try:
 os.environ['APP_SETTINGS']
except KeyError:
 os.environ['APP_SETTINGS'] = os.path.join(app.root_path, 'default-settings.py')

app.config.from_envvar('APP_SETTINGS'
database = SqliteDatabase(app.config['DATABASE'])
@app.before_request
def _db_connect():
  database.connect()

@app.teardown_request
def _db_close(exc):
  if not database.is_closed():
  database.close()

def init_db():
  for tbl in database.get_tables():
    database.execute_sql("drop table " + tbl)
    database.create_tables([table1, table2, ..., tableN])

with a default-settings.py file containing,

DATABASE='app_stuff.db'
SECRET_KEY='ApPlIcAtIoNsEcReTkEy'
TESTING=False

And this will work nicely with PeeWee as the ORM. The tricky bit has always been how to do the same with unittest.

A simple unittest script, say, thewebapp-test.py, might start with,

import os
import unittest
import tempfile
import sqlite3

class ThewebappTestCase(unittest.TestCase):

  def setUp(self):
    self.db_fd, thewebapp.app.config['DATABASE'] = tempfile.mkstemp()
    self.app = thewebapp.app.test_client()
      with thewebapp.app.app_context():
        thewebapp.init_db()

  def tearDown(self):
    os.unlink(thewebapp.app.config['DATABASE'])

with the following in a file, test-settings.py,

import tempfile

SECRET_KEY='TeStApPlIcAtIoNsEcReTkEy'
TESTING=True
db_fd, DATABASE = tempfile.mkstemp()

Then all we need to do to run the tests (so that a new temporary is created (and removed) for each test is simply set an environment variable when running the tests,

APP_SETTINGS="test-settings.py" python ./thewebapp-test.py

Now, I’ll be the first to admit that some of this still looks a bit hairy and there’s room for improvement but I am at least able to provide a reasonable set of tests for the application without hacks and special cases (not that Python does case. Grrr)

The goal of the next project is to find a way of using PeeWee to generate a dump of the database.

Pythonic balancer control

In my day job I work with a lot of services that sit behind a load balancer providing high availability across multiple backend hosts.

Now, inevitably there are times when these services or their hosting servers require maintenance. And sometimes we want to do some investigation and troubleshooting against a running service, but without it taking any live traffic; it simply isn’t practical to try and involve the network team’s assistance with disabling interfaces gracefully. The usual approach is to consider using a server-side firewall to block inbound port access but this can be clumsy and can actually impact live traffic, albeit briefly.

One solution I like is to use an intermediate monitor (or watchdog) service that provides a healthcheck URL for the load balancer, say, http://192.168.1.100/my-service/healthcheck, where the returned status is derived from the application ports being monitored.

Now, we want this to be as lightweight as possible, so we can choose something like Python’s Flask and uwsgi (or Ruby Sinatra) to provide a simple service listener like,

#!/usr/bin/env python
#
from flask import Flask, abort, request, Response, redirect
import os
import requests

app = Flask(__name__)

@app.route('/my-service/healthcheck', methods=["GET"])
def heartbeat():
 resp = Response(response = "OK", status = 200, content_type = "text/plain")

# Now check for the node statuses
 nodes = ( "inbound", "outbound", "stats" )
 for node in nodes:
 req = requests.get("http://localhost:7070/" + node + "/isalive")
 if(req.status_code != 200):
 resp.status = "FAILED" 
 resp.status_code = req.status_code

return(resp)

if __name__ == "__main__":
 app.run()

And obviously I have skipped the setup with pip, virtualenv and the like, but that’s routine enough.

The beauty with this kind of approach is that with a few extra lines before the service port polling we can spoof an outage and allow the load balancer to complete any existing client connections (that the firewall approach will prevent) while marking the node out of action,

 # Check for the maintenance file and signal graceful failure
 if(os.path.isfile("maintenance")):
 resp.status = "Under maintenance. Remove maintenance file when complete"
 resp.status_code = 503

Now, by simple touching a file called maintenance in the directory where the application is run from, the next poll from the load balancer will register the failure, and we can test this with cURL,

$ curl -v http://localhost:7070/my-service/healthcheck
* Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 7070 (#0)
> GET /my-service/healthcheck HTTP/1.1
> Host: localhost:7070
> User-Agent: curl/7.52.1
> Accept: */*
> 
< HTTP/1.1 503 SERVICE UNAVAILABLE
< Content-Type: text/plain
< Content-Length: 2

Remove the file and the traffic will flow again. Remote control of the load balancer without stopping any services, reboot persistent and allowing us time and space to investigate as we please.

Flask over https and local root CA

I recently had a week’s PTO and decided to spend the time getting to grips with setting up a private root CA so that I could try getting a python flask application to use client certificates (signed by the server) for authorization to request upload resources.

It was quite a struggle, more because of the root CA than anything else but since the aim was to gain an understanding of how this kind of set up worked, how to structure the requests and do the work in python (with a ruby version to follow no doubt) I am happy with the progress so far.

I will post a couple of pages describing the work I did with references and how I got past some of the major sticking points.