Source code for pythonie.blueprints.database

"""Example of using signals to manage a database connection

"""

import logging

import flask

import sqlalchemy

blueprint = flask.Blueprint("database", __name__)

# Set up the SQLALchemy table, this can happen elsewhere.
meta = sqlalchemy.MetaData()
table = sqlalchemy.Table("books", meta,
    sqlalchemy.Column("title", sqlalchemy.String(100)),
    sqlalchemy.Column("description", sqlalchemy.Text()),
)
engine = None  # A global reference to the engine, per-process.

@blueprint.before_app_first_request
def init_db():
    """Creates the initial database connection

    Fired before the first HTTP request (to any part of the site).

    """
    global engine
    engine = sqlalchemy.create_engine(flask.current_app.config["DATABASE"])
    meta.bind = engine
    meta.create_all()

@blueprint.before_request
def connect():
    """Creates a per request connection and transaction

    """
    flask.g.connection = engine.connect()
    flask.g.transaction = flask.g.connection.begin()

@blueprint.teardown_request
def disconnect(exception):
    """Commits or rolls back the transaction and disconnects

    """
    try:
        if exception is not None:
            flask.g.transaction.rollback()
        else:
            flask.g.transaction.commit()
        flask.g.connection.close()
    except Exception:
        logging.exception("Problem closing the DB")

@blueprint.route("/")
def index():
    """List all the books in JSON

    To see in action go to http://building-webapps-with-flask.herokuapp.com/database/

    """
    query = table.select()
    results = flask.g.connection.execute(query)
    return flask.jsonify({
        "books": [dict(row) for row in results]
    })

@blueprint.route("/add/", methods=["POST"])
def add():
    """Add a new book via POST

    To see in action go to http://building-webapps-with-flask.herokuapp.com/database/add/

    Note the use of methods in the decorator to only accept POST.

    :param title: The book's title
    :param description: The book's description

    """
    # Tip: Use something like wtforms or colander for validation
    query = table.insert().values(
        title=flask.request.form["title"],
        description=flask.request.form["description"]
    )
    flask.g.connection.execute(query)
    return flask.redirect(flask.url_for(".index"))