Most applications need to be backed up with a database. In this post we will start to add a SQLite database to the estimator application and use SQLAlchemy to access the data.

Part way through writing this post I realised that to integrate a database correctly, I will need to make different changes to allow the application to scale better.

I realised that I need to be able to run it in development, test and production modes. In development I can set up entities in the database for my ongoing development. In test I want to begin with a clean database at the start of the test run, set up only those cases that are required and clear it all down afterwards.

Running as production will require a different set up, perhaps a different database, for example MySQL, running on a different server.

So this post serves only as an introduction to SQLAlchemy using SQLite in a development mode. I can set up the database to allow the tests to pass, but since the database can be changed by interacting with the application, this is not the best overall solution. Consider it a small step in the right direction.

SQLAlchemy with SQLite

SQLAlchemy is an object relational mapping module that allows you to define Python objects, and interact with them, without writing any SQL.

SQLite is a self-contained database which is used by many small applications, including those running on mobile devices. It can be a good choice when there is not going to be significant amounts of data stored permanently.

In our application, the data does not really need to be kept much longer than the estimation session, so nothing will be lost by regularly purging older records. I may change this when moving to production, but for development and test, SQLite is the right choice.

We can add SQLAlchemy to the project by install the Flask extension Flask-SQLAlchemy. First activate the virtualenv and then run:

pip install Flask-SQLAlchemy

We should also update the requirements so that anybody else trying to run the code can keep up.

pip freeze >requirements.txt

Creating the database

Add an import and some lines of code to estimator/__init__.py to instantiate the database. Note that the app object is passed to the constructor.

from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///../data.sqlite'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

This is where I am having the issue. To define the database, I already need to have created the app. But I have not placed the application into test mode yet. That is done later by the Pytest fixture in tests\conftest.py but that is too late to set up the database differently for test mode. So I cannot set up the database with different options at this point.

The solution (coming in another post soon) is to place the application creation into a create_app function so that it can be passed different configuration settings.

Defining the tables

Rather than use SQL create statements, SQLAlchemy allows us to define the object and then it will create the table for you. It will name the columns based on the fields in the object. The table name or columns can be overridden, if necessary.

To get started, I am just going to create a single, simple table to hold the groups which will be identified by a name field. This will have a integer primary key and the group name.

To do so, I’ve created a subclass of the SQLAlchemy Model class. It provides various accessors manipulate the rows in the table. I anticipate several tables, so I have created a new database module (a sub folder with an empty __init__.py file in it) to hold them.

At the moment the following code is in a file called modules.py:

from estimator import db

class Group(db.Model):
	"""A group that team members can join and contains issues"""

	id = db.Column(db.Integer, primary_key=True)
	name = db.Column(db.String)

	def __init__(self, name):
		self.name = name

	def __repr__(self):
		return '<Estimation group: {}>'.format(self.name)

Before adding much more to the application, let’s test how this works using the Python interpreter. First import the required code and run the create_all method to initialise the database.

>>> from estimator import db
>>> from database.models import Group
>>> db.create_all()

Now we should be able to create a few group rows and commit the rows to our SQLite DB.

g = Group("NewGroup")
g2 = Group("SecondGroup")
db.session.add(g)
db.session.add(g2)
db.session.commit()

We can query back the rows like this:

>>> Group.query.first()
<Estimation group: NewGroup>
>>> 
>>> Group.query.filter_by(name='SecondGroup').first()
<Estimation group: SecondGroup>

The above output was how the __repr__ method returned the data. Later I will change this will to use JSON representation.

Updating the Tests

For now, I will simply seed some rows in the database to allow the tests to pass. I’m adding the following statements to the start of the tests/test_rest_controller.py file that will ensure the required test cases are in the database.

from estimator import db
from database.models import Group

# Initialize the in-memory DB and add some entities required for testing
db.create_all()
test_group = Group('TestGroup')
db.session.add(test_group)
existing_group = Group('GroupAlreadyExists')
db.session.add(existing_group)
db.session.commit()

Let’s update the test_create_group method. Since writing the test originally I have read that the response from a create with POST should be empty and that the HTTP header should contain the location of the resource, rather than receiving and data back in the response.

The other change here is to read the group back from the database based on the name field and verify that it was found.

def test_create_group(client):
	response = client.post('/rest/v1/group/NewGroup')
	assert response.status == '201 CREATED'
	assert response.get_data() == '{}'
	assert response.headers['Location'] == 'http://localhost/rest/v1/group/NewGroup'
	assert response.headers['Content-Type'] == 'application/json'
	g = Group.query.filter_by(name='NewGroup').first()
	assert g

Updating the controller

We need to import a few new things in rest_controller.py. Building a URL to return when the group is created is best done using Flask’s url_for method. We also need to import the db object and the models

from flask import jsonify, Response, url_for
from estimator import app, db
from database.models import Group

The query_group method that responds to GET requests will query the database and return a simple JSON object. If the group is not found, we return the HTTP code 404.

@app.route('/rest/v1/group/<groupname>', methods = ['GET'])
def query_group(groupname):
	group = Group.query.filter_by(name=groupname).first()
	if group:
		return jsonify({ "groupname" : group.name })
	else:
		return jsonify({ "message" : "Group not found"}), 404, {'mimetype': 'application/json'}

For the create_group method first checks that the group is not already in the database, returning a HTTP 400 error if found. If not, it will create the group, persist it to the database and return a response that includes a header containing the location.

@app.route('/rest/v1/group/<groupname>', methods = ['POST'])
def create_group(groupname):
	group = Group.query.filter_by(name=groupname).first()
	if group:
		body = { "message" : "Group already exists" }
		code = 400
		return jsonify(body), code
	else:
		print("Creating new group {}".format(groupname))
		group = Group(groupname)
		db.session.add(group)
		db.session.commit()
		body = {}
		code = 201
		return jsonify({}), code, {'location': url_for('query_group', groupname=groupname)}

Clean up

The tests should pass when run now, however, only the first time. Because I am using a file based database file, when the test is run for the second time, the NewGroup is already there. In my next post I will be cleaning this up, but for now the following hack will do.

Before running the tests, I will search for and delete any rows matching this group name:

# clear out any existing groups
to_delete = Group.query.filter_by(name='NewGroup')
[db.session.delete(item) for item in to_delete]
db.session.commit()

I don’t want to check in the database file. So I will add this line to the .gitignore file:

data.sqlite

I have also created a file, dbinit.py, to initialise the database. It only has to be run once to create the database tables.

from estimator import db
from database.models import *
db.create_all()