app.py を起動し、データベースを強制終了してヒットすると、次のようになります/api/foo。
peewee.OperationalError: サーバーに接続できませんでした: 接続が拒否されました
データベースを元に戻すと、次のようになり/api/fooます。
peewee.OperationalError: 管理者コマンドにより接続を終了しています\nSSL 接続が予期せず閉じられました\n
もう一度押すと、次の/api/fooようになります。
peewee.InterfaceError: 接続は既に閉じられています
テストケース
test_case/__init__.py
#!/usr/bin/env python
from os import environ
from bottle import Bottle, request, response
from playhouse.db_url import connect
bottle_api = Bottle()
db = connect(environ['RDBMS_URI'])
from test_case.foo.models import Foo
db.connect() # Not needed, but do want to throw errors ASAP
db.create_tables([Foo], safe=True) # Create tables (if they don't exist)
from test_case.foo.routes import foo_api
bottle_api.merge(foo_api)
bottle_api.catchall = False
@bottle_api.hook('before_request')
def _connect_db():
print 'Connecting to db'
db.connect()
@bottle_api.hook('after_request')
def _close_db():
print 'Closing db'
if not db.is_closed():
db.close()
def error_catcher(environment, start_response):
try:
return bottle_api.wsgi(environment, start_response)
except Exception as e:
environment['PATH_INFO'] = '/api/error'
environment['api_error'] = e
return bottle_api.wsgi(environment, start_response)
@bottle_api.route('/api/error')
def global_error():
response.status = 500
return {'error': (lambda res: res[res.find("'") + 1:res.rfind("'")])(
str(request.environ['api_error'].__class__)),
'error_message': request.environ['api_error'].message}
test_case/__main__.py
from __init__ import bottle_api
# Or `from __init__ import bottle_api`; `from bottle import run`;
# Then `run(error_catcher, port=5555)`
bottle_api.run(port=5555)
test_case/foo/__init__.py
test_case/foo/models.py
from peewee import Model, CharField
from test_case import db
class Foo(Model):
id = CharField(primary_key=True)
class Meta(object):
database = db
test_case/foo/routes.py
from bottle import Bottle
from playhouse.shortcuts import model_to_dict
from test_case.foo.models import Foo
foo_api = Bottle()
@foo_api.get('/api/foo')
def retrieve_foos():
return {'foos': tuple(model_to_dict(foo) for foo in Foo.select())}