Directory layout

.
├── app.py
├── migrations
│   ├── 001.rollback.sql
│   └── 001.sql
├── model.py
├── templates
│   └── counter.html
└── test_app.py

The application code

examples/flask/sqlite-counter-yoyo/app.py

from flask import Flask, render_template, request, abort
import logging
import os
from model import apply_db_migrations, setup_db, create_classes

def create_app():
    app = Flask(__name__)
    app.logger.setLevel(logging.INFO)

    with app.app_context():
        db_file = os.environ.get('COUNT_DB') or 'counter.db'
        db_uri = 'sqlite:///' + db_file
        app.config['SQLALCHEMY_DATABASE_URI'] = db_uri
        app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
        apply_db_migrations(db_uri)
        db = setup_db(app)
        Counter = create_classes(db)

        @app.route('/')
        @app.route('/<name>')
        def counter(name = None):
            if name:
                if name == 'favicon.ico':
                    return abort(404)
                app.logger.info("name %s", name)
                cntr = db.session.query(Counter).filter_by(name=name).first()
                app.logger.info(cntr)
                if not cntr:
                    cntr = Counter(name=name, count=1)
                    db.session.add(cntr)
                else:
                    cntr.count += 1
                db.session.commit()
                app.logger.info(cntr)

                return render_template('counter.html', name = name, counter = cntr.count)
            else:
                counters = [ {"name": counter.name, "count" : counter.count} for counter in db.session.query(Counter).all()]
                app.logger.info(counters)
                return render_template('counter.html', counters = counters)



    return app

The way we run queries changed. We now need to call db.session.query and pass it the class representing the table.

The SQL migration scripts

examples/flask/sqlite-counter-yoyo/migrations/001.sql

CREATE TABLE counter (
	name VARCHAR(80) NOT NULL,
	count INTEGER NOT NULL,
	PRIMARY KEY (name),
	UNIQUE (name)
);

examples/flask/sqlite-counter-yoyo/migrations/001.rollback.sql

DROP TABLE counter;

The SQLAlchemy code

examples/flask/sqlite-counter-yoyo/model.py

from flask_sqlalchemy import SQLAlchemy
from yoyo import read_migrations
from yoyo import get_backend
from sqlalchemy.ext.automap import automap_base

def apply_db_migrations(db_uri):
    backend = get_backend(db_uri)
    migrations = read_migrations('./migrations')
    with backend.lock():
        backend.apply_migrations(backend.to_apply(migrations))


def setup_db(app):
    db = SQLAlchemy()
    db.init_app(app)
    return db

def create_classes(db):
    Base = automap_base()
    Base.prepare(db.engine, reflect=True)
    Counter = Base.classes.counter
    return Counter

There are two ways to use SQLAlchemy. Either we declare the details of the schema in Python statements as we did in the other version or we use the automap feature of SQLAlchemy.

As we already have our SQL declarations in the migration files I thought it would be better to try to use the automap_base.

The Jinja template

examples/flask/sqlite-counter-yoyo/templates/counter.html

<h1>Counter</h1>
{% if name %}
  {{ name }} : {{ counter}}
{% else %}
    {% if counters %}
        <ul>
        {% for counter in counters %}
            <li><a href="/{{counter.name}}">{{counter.name}} {{counter.count}}</a></li>
        {% endfor %}
        </ul>
    
    {% else %}
        Try <a href="/something">something</a>
    {% endif %}
{% endif %}

The tests

examples/flask/sqlite-counter-yoyo/test_app.py

from app import create_app
import os

def test_app(tmpdir):
    os.environ['COUNT_DB'] = str(os.path.join(tmpdir, "test.db"))
    print(os.environ['COUNT_DB'])
    web = create_app().test_client()

    rv = web.get('/')
    assert rv.status == '200 OK'
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    assert '<a href="/something">something</a>' in rv.data.decode('utf-8')

    rv = web.get('/blue')
    assert rv.status == '200 OK'
    #print(rv.data)
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    assert 'blue : 1' in rv.data.decode('utf-8')
    rv = web.get('/')
    assert rv.status == '200 OK'
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    assert '<a href="/blue">blue 1</a>' in rv.data.decode('utf-8')
    assert 'something' not in rv.data.decode('utf-8')


    rv = web.get('/blue')
    assert rv.status == '200 OK'
    #print(rv.data)
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    assert 'blue : 2' in rv.data.decode('utf-8')
    assert 'red' not in rv.data.decode('utf-8')
    assert 'something' not in rv.data.decode('utf-8')


    rv = web.get('/red')
    assert rv.status == '200 OK'
    #print(rv.data)
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    #assert 'blue : 2' in rv.data.decode('utf-8')
    assert 'red : 1' in rv.data.decode('utf-8')
    rv = web.get('/')
    assert rv.status == '200 OK'
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    assert '<a href="/blue">blue 2</a>' in rv.data.decode('utf-8')
    assert '<a href="/red">red 1</a>' in rv.data.decode('utf-8')
    assert 'something' not in rv.data.decode('utf-8')

# Verify that the two test-cases indeed use different databases
def test_app_clean_db(tmpdir):
    os.environ['COUNT_DB'] = str(os.path.join(tmpdir, "test.db"))
    print(os.environ['COUNT_DB'])
    web = create_app().test_client()

    rv = web.get('/')
    assert rv.status == '200 OK'
    assert '<h1>Counter</h1>' in rv.data.decode('utf-8')
    assert '<a href="/something">something</a>' in rv.data.decode('utf-8')
    assert 'blue' not in rv.data.decode('utf-8')
    assert 'red' not in rv.data.decode('utf-8')