Tutorial
A simple application
Let’s start with an ‘hello, world’ example. It’s a small application for manipulating an SQLite database of counter. Here’s the schema:
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.
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, execute, query, query_column, query_value, transactional
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:
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:
dbkit also makes dealing with transactions very easy. It provides two mechanisms: the dbkit.transaction context manager and dbkit.transactional decorator. Let’s implement incrementing the counter using the context manager:
def increment_counter(counter, by):
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:
@transactional
def delete_counter(counter):
"""
Delete a counter.
"""
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(f"{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=DictFactory,
)
dbkit.DictFactory is a row factory that generates a result set where each row is a dictionary. The default row factory is dbkit.TupleFactory, which yields tuples for each row in the result set. Using dbkit.DictFactory, we’d print the counters and values like so:
def print_counters_and_values():
for row in dump_counters_dict():
print(f"{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(f"Syntax: {path.basename(filename)} {cmds} [args]", file=dest)
Bingo! You now has a simple counter manipulation tool.
To do
Connection pools.