1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
"""
Bottle-sqlite is a plugin that integrates SQLite3 with your Bottle
application. It automatically connects to a database at the beginning of a
request, passes the database handle to the route callback and closes the
connection afterwards.
To automatically detect routes that need a database connection, the plugin
searches for route callbacks that require a `db` keyword argument
(configurable) and skips routes that do not. This removes any overhead for
routes that don't need a database connection.
Usage Example::
import bottle
from bottle.ext import sqlite
app = bottle.Bottle()
plugin = sqlite.Plugin(dbfile='/tmp/test.db')
app.install(plugin)
@app.route('/show/:item')
def show(item, db):
row = db.execute('SELECT * from items where name=?', item).fetchone()
if row:
return template('showitem', page=row)
return HTTPError(404, "Page not found")
"""
__author__ = "Marcel Hellkamp"
__version__ = "0.2.0"
__license__ = "MIT"
### CUT HERE (see setup.py)
import sqlite3
import inspect
import bottle
# PluginError is defined to bottle >= 0.10
if not hasattr(bottle, "PluginError"):
class PluginError(bottle.BottleException):
pass
bottle.PluginError = PluginError
class SQLitePlugin(object):
"""This plugin passes an sqlite3 database handle to route callbacks
that accept a `db` keyword argument. If a callback does not expect
such a parameter, no connection is made. You can override the database
settings on a per-route basis."""
name = "sqlite"
api = 2
""" python3 moves unicode to str """
try:
unicode
except NameError:
unicode = str
def __init__(
self,
dbfile=":memory:",
autocommit=True,
dictrows=True,
keyword="db",
text_factory=unicode,
functions=None,
aggregates=None,
collations=None,
extensions=None,
):
self.dbfile = dbfile
self.autocommit = autocommit
self.dictrows = dictrows
self.keyword = keyword
self.text_factory = text_factory
self.functions = functions or {}
self.aggregates = aggregates or {}
self.collations = collations or {}
self.extensions = extensions or ()
def setup(self, app):
"""Make sure that other installed plugins don't affect the same
keyword argument."""
for other in app.plugins:
if not isinstance(other, SQLitePlugin):
continue
if other.keyword == self.keyword:
raise PluginError(
"Found another sqlite plugin with "
"conflicting settings (non-unique keyword)."
)
elif other.name == self.name:
self.name += "_%s" % self.keyword
def apply(self, callback, route):
# hack to support bottle v0.9.x
if bottle.__version__.startswith("0.9"):
config = route["config"]
_callback = route["callback"]
else:
config = route.config
_callback = route.callback
# Override global configuration with route-specific values.
if "sqlite" in config:
# support for configuration before `ConfigDict` namespaces
g = lambda key, default: config.get("sqlite", {}).get(key, default)
else:
g = lambda key, default: config.get("sqlite." + key, default)
dbfile = g("dbfile", self.dbfile)
autocommit = g("autocommit", self.autocommit)
dictrows = g("dictrows", self.dictrows)
keyword = g("keyword", self.keyword)
text_factory = g("text_factory", self.text_factory)
functions = g("functions", self.functions)
aggregates = g("aggregates", self.aggregates)
collations = g("collations", self.collations)
extensions = g("extensions", self.extensions)
# Test if the original callback accepts a 'db' keyword.
# Ignore it if it does not need a database handle.
argspec = inspect.getfullargspec(_callback)
if keyword not in argspec.args:
return callback
def wrapper(*args, **kwargs):
# Connect to the database
db = sqlite3.connect(dbfile)
# set text factory
db.text_factory = text_factory
# This enables column access by name: row['column_name']
if dictrows:
db.row_factory = sqlite3.Row
# Create user functions, aggregates and collations
for name, value in functions.items():
db.create_function(name, *value)
for name, value in aggregates.items():
db.create_aggregate(name, *value)
for name, value in collations.items():
db.create_collation(name, value)
for name in extensions:
db.enable_load_extension(True)
db.execute("SELECT load_extension(?)", (name,))
db.enable_load_extension(False)
# Add the connection handle as a keyword argument.
kwargs[keyword] = db
try:
rv = callback(*args, **kwargs)
if autocommit:
db.commit()
except sqlite3.IntegrityError as e:
db.rollback()
raise bottle.HTTPError(500, "Database Error", e)
except bottle.HTTPError as e:
raise
except bottle.HTTPResponse as e:
if autocommit:
db.commit()
raise
finally:
db.close()
return rv
# Replace the route callback with the wrapped one.
return wrapper
Plugin = SQLitePlugin
|
