Framework Integration

For web applications, it is common to open a connection when a request is received, and to close the connection when the response is delivered. In this section I will describe how to add hooks to your web app to ensure the database connection is handled properly.

These steps will ensure that regardless of whether you’re using a simple SQLite database, or a pool of multiple Postgres connections, peewee will handle the connections correctly.

Note

Applications that receive lots of traffic may benefit from using a connection pool to mitigate the cost of setting up and tearing down connections on every request.

Flask

Flask and peewee are a great combo and my go-to for projects of any size. Flask provides two hooks which we will use to open and close our db connection. We’ll open the connection when a request is received, then close it when the response is returned.

from flask import Flask
from peewee import *
database = SqliteDatabase('my_app.db')
app = Flask(__name__)
# This hook ensures that a connection is opened to handle any queries
# generated by the request.
@app.before_request
def _db_connect():
    database.connect()
# This hook ensures that the connection is closed when we've finished
# processing the request.
@app.teardown_request
def _db_close(exc):
    if not database.is_closed():
        database.close()

Django

While it’s less common to see peewee used with Django, it is actually very easy to use the two. To manage your peewee database connections with Django, the easiest way in my opinion is to add a middleware to your app. The middleware should be the very first in the list of middlewares, to ensure it runs first when a request is handled, and last when the response is returned.

If you have a django project named my_blog and your peewee database is defined in the module my_blog.db, you might add the following middleware class:

# middleware.py
from my_blog.db import database  # Import the peewee database instance.
def PeeweeConnectionMiddleware(get_response):
    def middleware(request):
        database.connect()
        try:
            response = get_response(request)
        finally:
            if not database.is_closed():
                database.close()
        return response
    return middleware
# Older Django < 1.10 middleware.
class PeeweeConnectionMiddleware(object):
    def process_request(self, request):
        database.connect()
    def process_response(self, request, response):
        if not database.is_closed():
            database.close()
        return response

To ensure this middleware gets executed, add it to your settings module:

# settings.py
MIDDLEWARE_CLASSES = (
    # Our custom middleware appears first in the list.
    'my_blog.middleware.PeeweeConnectionMiddleware',
    # These are the default Django 1.7 middlewares. Yours may differ,
    # but the important this is that our Peewee middleware comes first.
    'django.middleware.common.CommonMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
)
# ... other Django settings ...

Bottle

I haven’t used bottle myself, but looking at the documentation I believe the following code should ensure the database connections are properly managed:

# app.py
from bottle import hook  #, route, etc, etc.
from peewee import *
db = SqliteDatabase('my-bottle-app.db')
@hook('before_request')
def _connect_db():
    db.connect()
@hook('after_request')
def _close_db():
    if not db.is_closed():
        db.close()
# Rest of your bottle app goes here.

Web.py

See the documentation for application processors.

db = SqliteDatabase('my_webpy_app.db')
def connection_processor(handler):
    db.connect()
    try:
        return handler()
    finally:
        if not db.is_closed():
            db.close()
app.add_processor(connection_processor)

Tornado

It looks like Tornado’s RequestHandler class implements two hooks which can be used to open and close connections when a request is handled.

from tornado.web import RequestHandler
db = SqliteDatabase('my_db.db')
class PeeweeRequestHandler(RequestHandler):
    def prepare(self):
        db.connect()
        return super(PeeweeRequestHandler, self).prepare()
    def on_finish(self):
        if not db.is_closed():
            db.close()
        return super(PeeweeRequestHandler, self).on_finish()

In your app, instead of extending the default RequestHandler, now you can extend PeeweeRequestHandler.

Note that this does not address how to use peewee asynchronously with Tornado or another event loop.

Wheezy.web

The connection handling code can be placed in a middleware.

def peewee_middleware(request, following):
    db.connect()
    try:
        response = following(request)
    finally:
        if not db.is_closed():
            db.close()
    return response
app = WSGIApplication(middleware=[
    lambda x: peewee_middleware,
    # ... other middlewares ...
])

Thanks to GitHub user @tuukkamustonen for submitting this code.

Falcon

The connection handling code can be placed in a middleware component.

import falcon
from peewee import *
database = SqliteDatabase('my_app.db')
class PeeweeConnectionMiddleware(object):
    def process_request(self, req, resp):
        database.connect()
    def process_response(self, req, resp, resource, req_succeeded):
        if not database.is_closed():
            database.close()
application = falcon.API(middleware=[
    PeeweeConnectionMiddleware(),
    # ... other middlewares ...
])

Pyramid

Set up a Request factory that handles database connection lifetime as follows:

from pyramid.request import Request
db = SqliteDatabase('pyramidapp.db')
class MyRequest(Request):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        db.connect()
        self.add_finished_callback(self.finish)
    def finish(self, request):
        if not db.is_closed():
            db.close()

In your application main() make sure MyRequest is used as request_factory:

def main(global_settings, **settings):
    config = Configurator(settings=settings, ...)
    config.set_request_factory(MyRequest)

CherryPy

See Publish/Subscribe pattern.

def _db_connect():
    db.connect()
def _db_close():
    if not db.is_closed():
        db.close()
cherrypy.engine.subscribe('before_request', _db_connect)
cherrypy.engine.subscribe('after_request', _db_close)

Sanic

In Sanic, the connection handling code can be placed in the request and response middleware sanic middleware.

# app.py
@app.middleware('request')
async def handle_request(request):
    db.connect()
@app.middleware('response')
async def handle_response(request, response):
    if not db.is_closed():
        db.close()

FastAPI

Similar to Flask, FastAPI provides two event based hooks which we will use to open and close our db connection. We’ll open the connection when a request is received, then close it when the response is returned.

from fastapi import FastAPI
from peewee import *
db = SqliteDatabase('my_app.db')
app = FastAPI()
# This hook ensures that a connection is opened to handle any queries
# generated by the request.
@app.on_event("startup")
def startup():
    db.connect()
# This hook ensures that the connection is closed when we've finished
# processing the request.
@app.on_event("shutdown")
def shutdown():
    if not db.is_closed():
        db.close()

Other frameworks

Don’t see your framework here? Please open a GitHub ticket and I’ll see about adding a section, or better yet, submit a documentation pull-request.