Category Archives: PeeWee

Dynamic (in-memory) zip file creation

With a recent project I wanted to offer a download of a bunch of files as a ZIP download to make redistribution. I started by thinking that this would involve creating a zip file on disk and then serving it to the client.

It doesn’t.

This method creates an in-memory zip image of a temporary directory in which the required file are created. The directory is deleted before the zip file is returned to the client. Note the use of send_file to return the content and how it allows us to specify the name of the downloaded file.

If running the app under uwsgi there are some precautions to take:

  • import the send_file module from flask (the default wsgi version is broken)
  • Include ‘wsgi-disable-file-wrapper’ in the uwsgi settings
import os
import io
import tempfile
import zipfile
from peewee import *

# A simple function to write a file in a given directory
def makeFile(dir, filename, data):
    fname = str(dir) + '/' + str(filename)
    fh = open(fname, "w")
    fh.write(data)
    fh.close()
    return True
"""
Function to create a zip file containing the application configuration files
To create the zip file, we need to create a memory file (BytesIO))
Params:
 row: Database row object containing columns for filename and
 content to be written in the form
 { {"file1": "Content for file 1}, {"file2": "Different content for file"} }
Returns:
 data: BytesIO object containing the zipped files
"""
def mkZipFile(row):
    zipdir = tempfile.mkdtemp(prefix='/tmp/')
    oldpath = os.getcwd()
    os.chdir(zipdir)

    jdata = json.loads(row)
    for conf in jdata:
        for f in conf:
            makeFile(zipdir, f, conf[f])

    # Create the in-memory zip image
    data = io.BytesIO()
    with zipfile.ZipFile(data, mode='w') as z:
        for fname in os.listdir("."):
            z.write(fname)
            os.unlink(fname)
    data.seek(0)

    os.chdir(oldpath)
    os.rmdir(zipdir)
    return data

The route takes the form

"""
Route to create a zip file containing all the required data files
Params:
 id: the numeric id of the record in the database
Returns:
 ZIP: file in ZIP format containing the download files
"""
@app.route('/download/<int:id>.zip', methods=['GET'])
def showZip(id):
    rs = db_table.select(db_table.data_column).where(db_table.id == id).get()

    zfile = mkZipFile(rs)
    if not zfile:
        return Response(response="Invalid zip file", status="400")

    return send_file(
        zfile,
        mimetype='application/zip',
        as_attachment=True,
        attachment_filename='file_bundle.zip')

References

Advertisements

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.