Let’s start with an ‘hello, world’ example. It’s a small application for manipulating an SQLite database of counter. Here’s the schema:
CREATE TABLE counters (
counter TEXT PRIMARY KEY,
value INTEGER
);
You’ll find that file in the examples directory, and it’s called counters.sql. Let’s create the database:
$ sqlite3 counters.sqlite < counters.sql
You should now have the database set up.
You’ll find the file we’ll be working on in the examples as counters.py.
Now let’s import some of the libraries we’ll be needing for this project:
from contextlib import closing
from os import path
import sqlite3
import sys
from dbkit import connect, transaction, \
There are a few different thing we want to be able to do to the counter, such as setting a counter, deleting a counter, listing counters, incrementing a counter, and getting the value of a counter. We’ll need to implement the code to do those.
One of the neat things about dbkit is that you don’t have to worry about passing around database connections. Instead, you create a context in which the queries are ran, and dbkit itself does the work. Thus, we can do something like this:
value = query_value(
'SELECT value FROM counters WHERE counter = ?',
(counter,),
default=0)
And we don’t need to worry about the database connection we’re actually dealing with. With that in mind, here’s how we’d implement getting a counter’s value with dbkit.query_value():
def get_counter(counter):
"""Get the value of a counter."""
print query_value(
'SELECT value FROM counters WHERE counter = ?',
(counter,),
default=0)
To perform updates, there’s the dbkit.execute() function. Here’s how we increment a counter’s value:
def set_counter(counter, value):
"""Set a counter."""
with transaction():
execute(
'REPLACE INTO counters (counter, value) VALUES (?, ?)',
(counter, value))
dbkit also makes dealing with transactions very easy. It provides two mechanisms: the dbkit.transaction() context manager, as demonstrated above, and dbkit.transactional() decorator. Let’s implement incrementing the counter using the context manager:
def increment_counter(counter, by):
"""Modify the value of a counter by a certain amount."""
with transaction():
execute(
'UPDATE counters SET value = value + ? WHERE counter = ?',
(by, counter))
With the decorator, we’d write the function like so:
@transactional
def increment_counter(counter, by):
execute(
'UPDATE counters SET value = value + ? WHERE counter = ?',
(by, counter))
Both are useful in different circumstances.
Deleting a counter:
def delete_counter(counter):
"""Delete a counter."""
with transaction():
execute(
'DELETE FROM counters WHERE counter = ?',
(counter,))
dbkit also has ways to query for result sets. Once of these is dbkit.query_column(), which returns an iterable of the first column in the result set. Thus, to get a list of counters, we’d do this:
def list_counters():
"""List the names of all the stored counters."""
print "\n".join(query_column('SELECT counter FROM counters'))
One last thing that our tool ought to be able to do is dump the contents of the counters table. To do this, we can use dbkit.query():
def dump_counters():
"""Query the database for all counters and their values."""
return query('SELECT counter, value FROM counters')
This will return a sequence of result set rows you can iterate over like so:
def print_counters_and_values():
"""List all the counters and their values."""
for counter, value in dump_counters():
print "%s: %d" % (counter, value)
By default, query() will use tuples for each result set row, but if you’d prefer dictionaries, all you have to do is pass in a different row factory when you call dbkit.query() using the factory parameter:
def dump_counter_dict():
return query(
'SELECT counter, value FROM counters',
factory=dict_set)
dbkit.dict_set() is a row factory that generates a result set where each row is a dictionary. The default row factory is dbkit.tuple_set(), which yields tuples for each row in the result set. Using dbkit.dict_set(), we’d print the counters and values like so:
def print_counters_and_values():
for row in dump_counters_dict():
print '%s: %d' % (row['counter'], row['value'])
Now we have enough for our counter management application, so lets start on the main function. We’ll have the following subcommands: set, get, del, list, incr, list, and dump. The dispatch() function below deals with calling the right function based on the command line arguments, so all we need to create a database connection context with dbkit.connect(). It takes the database driver module as its first argument, and any parameters you’d pass to that module’s connect() function to create a new connection as its remaining arguments:
def main():
# This table tells us the subcommands, the functions to dispatch to,
# and their signatures.
command_table = {
'set': (set_counter, str, int),
'del': (delete_counter, str),
'get': (get_counter, str),
'list': (list_counters,),
'incr': (increment_counter, str, int),
'dump': (print_counters_and_values,),
}
with connect(sqlite3, 'counters.sqlite') as ctx:
with closing(ctx):
dispatch(command_table, sys.argv)
Finally, two utility methods, the first of which decides which of the functions to call based on a command dispatch table and the arguments the program was ran with:
def dispatch(table, args):
"""Dispatches to a function based on the contents of `args`."""
# No arguments: print help.
if len(args) == 1:
print_help(args[0], table)
sys.exit(0)
# Bad command or incorrect number of arguments: print help to stderr.
if args[1] not in table or len(args) != len(table[args[1]]) + 1:
print_help(args[0], table, dest=sys.stderr)
sys.exit(1)
# Cast all the arguments to fit their function's signature to ensure
# they're correct and to make them safe for consumption.
sig = table[args[1]]
try:
fixed_args = [type_(arg) for arg, type_ in zip(args[2:], sig[1:])]
except TypeError:
# If any are wrong, complain to stderr.
print_help(args[0], table, dest=sys.stderr)
sys.exit(1)
# Dispatch the call to the correct function.
sig[0](*fixed_args)
And a second for displaying help:
def print_help(filename, table, dest=sys.stdout):
"""Print help to the given destination file object."""
cmds = '|'.join(sorted(table.keys()))
print >> dest, "Syntax: %s %s [args]" % (path.basename(filename), cmds)
Bingo! You now has a simple counter manipulation tool.