TechEmpower.FrameworkBenchmarks/frameworks/Python/flask/app.py

#!/usr/bin/env python
from collections import namedtuple
from operator import attrgetter
from random import randint
import sys

from flask import Flask, request, render_template, make_response, jsonify
from pony import orm

if sys.version_info[0] == 3:
    xrange = range
_is_pypy = hasattr(sys, "pypy_version_info")
if _is_pypy:
    from psycopg2cffi import compat

    compat.register()

DBDRIVER = "postgres"
DBHOST = "tfb-database"

# setup

app = Flask(__name__)
app.config[
    "STORM_DATABASE_URI"
] = "{DBDRIVER}://benchmarkdbuser:benchmarkdbpass@{DBHOST}:5432/hello_world".format(
    DBDRIVER=DBDRIVER, DBHOST=DBHOST
)
app.config["JSONIFY_PRETTYPRINT_REGULAR"] = False
db = orm.Database()
db.bind(
    DBDRIVER,
    host=DBHOST,
    port=5432,
    user="benchmarkdbuser",
    password="benchmarkdbpass",
    database="hello_world",
)


class World(db.Entity):
    _table_ = "world"
    id = orm.PrimaryKey(int)
    randomNumber = orm.Required(int, column="randomnumber")

    def to_dict(self):
        """Return object data in easily serializeable format"""
        return {"id": self.id, "randomNumber": self.randomNumber}


class Fortune(db.Entity):
    _table_ = "fortune"
    id = orm.PrimaryKey(int, auto=True)
    message = orm.Required(str)


db.generate_mapping(create_tables=False)


def get_num_queries():
    try:
        num_queries = request.args.get("queries", 1, type=int)
    except ValueError:
        num_queries = 1
    if num_queries < 1:
        return 1
    if num_queries > 500:
        return 500
    return num_queries


def generate_ids(num_queries):
    ids = {randint(1, 10000) for _ in xrange(num_queries)}
    while len(ids) < num_queries:
        ids.add(randint(1, 10000))
    return list(sorted(ids))


@app.route("/json")
def hello():
    return jsonify(message="Hello, World!")


@app.route("/query")
def get_random_world():
    with orm.db_session(serializable=False):
        worlds = [World[ident].to_dict() for ident in generate_ids(get_num_queries())]
    return jsonify(worlds)


@app.route("/db")
def get_random_world_single():
    wid = randint(1, 10000)
    with orm.db_session(serializable=False):
        world = World[wid]
    return jsonify(world.to_dict())


@app.route("/fortunes")
def get_fortunes():
    with orm.db_session(serializable=False):
        fortunes = list(orm.select(fortune for fortune in Fortune))
    tmp_fortune = namedtuple("Fortune", ["id", "message"])
    fortunes.append(
        tmp_fortune(id=0, message="Additional fortune added at request time.")
    )
    fortunes.sort(key=attrgetter("message"))
    return render_template("fortunes.html", fortunes=fortunes)


@app.route("/updates")
def updates():
    """Test 5: Database Updates"""
    num_queries = get_num_queries()
    ids = generate_ids(num_queries)
    ids.sort()
    worlds = []
    with orm.db_session(serializable=False):
        for ident in ids:
            world = World[ident]
            world.randomNumber = randint(1, 10000)
            worlds.append({"id": world.id, "randomNumber": world.randomNumber})
    return jsonify(worlds)


@app.route("/plaintext")
def plaintext():
    """Test 6: Plaintext"""
    response = make_response(b"Hello, World!")
    response.content_type = "text/plain"
    return response


try:
    import meinheld

    meinheld.server.set_access_logger(None)
    meinheld.set_keepalive(120)
except ImportError:
    pass

# entry point for debugging
if __name__ == "__main__":
    app.run(debug=True)