from collections import namedtuple
from operator import itemgetter
from functools import partial
__all__ = [
'fetch_none',
'fetch_optional_value',
'fetch_required_value',
'fetch_values',
'fetch_optional_tuple',
'fetch_required_tuple',
'fetch_tuples',
'execute_none',
'execute_optional_value',
'execute_required_value',
'execute_values',
'execute_optional_tuple',
'execute_required_tuple',
'execute_tuples',
'callproc_none',
'callproc_optional_value',
'callproc_required_value',
'callproc_values',
'callproc_optional_tuple',
'callproc_required_tuple',
'callproc_tuples',
'make_execute_none',
'make_execute_optional_value',
'make_execute_required_value',
'make_execute_values',
'make_execute_optional_tuple',
'make_execute_required_tuple',
'make_execute_tuples',
'make_callproc_none',
'make_callproc_optional_value',
'make_callproc_required_value',
'make_callproc_values',
'make_callproc_optional_tuple',
'make_callproc_required_tuple',
'make_callproc_tuples',
]
def value_converter (row):
"""The standard converter for a single-column row.
This simply unpacks the single value from the row and returns it.
"""
(result,) = row
return result
def make_tuple_converter (cursor, typename=None):
"""Make a converter to convert a row from a cursor into a namedtuple
This starts by defining a namedtuple type based on the column names in the
cursor's current result set. Then the tuple converter is defined simply
to pass all the elements of the parameter row into the namedtuple type
constructor.
"""
columns = list(map (itemgetter (0), cursor.description))
if typename is None:
typename = '_'.join (columns)
nt = namedtuple (typename, columns)
def tuple_converter (row):
return nt (*row)
return tuple_converter
[docs]def fetch_none (cursor):
"""Fetch nothing from the cursor.
The result will be None. This is for running statements that do not
produce a result. In future this may be modified to verify that no
result is produced and complain if a result is produced.
"""
return None
def fetch_result (cursor, convert=None, required=False):
"""Fetch a single result from the cursor.
Fetches a single row from the cursor and converts it using the supplied
converter (or the identity function), and returns the converted row.
If there are no rows, then if required is True, an error is raised,
otherwise None is returned. In any case an error is raised if there
are more than one rows.
"""
result = cursor.fetchone ()
if result is None:
if required:
raise ValueError ("No result rows")
else:
return None
if cursor.fetchone () is not None:
raise ValueError ("Multiple result rows")
if convert is not None:
result = convert (result)
return result
def fetch_results (cursor, convert=None):
"""Fetch multiple results (zero, one, or more) from the cursor.
Fetches all rows from the cursor, converting them using the supplied
converter (or the identity function), and returns a list of the converted
rows.
"""
return list(map (convert, cursor.fetchall ()))
def fetch_value (cursor, required):
"""Fetch a single value result from the cursor.
The result must consist of at most one row containing exactly one field.
If required is True, there must be exactly one row. If required is False,
then None will be returned if there are no rows.
"""
return fetch_result (cursor, convert=value_converter, required=required)
fetch_optional_value = partial (fetch_value, required=False)
fetch_required_value = partial (fetch_value, required=True)
[docs]def fetch_values (cursor):
"""Fetch a multiple-value result from the cursor.
The result must consist of rows containing exactly one field.
"""
return fetch_results (cursor, convert=value_converter)
def fetch_tuple (cursor, required):
"""Fetch a single tuple result from the cursor.
The result must consist of at most one row. If required is True, there
must be exactly one row. If required is False, then None will be returned
if there are no rows. If there is a single row, the result will be a
namedtuple instance.
"""
return fetch_result (cursor,
convert=make_tuple_converter (cursor), required=required)
fetch_optional_tuple = partial (fetch_tuple, required=False)
fetch_required_tuple = partial (fetch_tuple, required=True)
[docs]def fetch_tuples (cursor):
"""Fetch a multiple-tuple result from the cursor.
The result will be a list of namedtuple instances.
"""
return fetch_results (cursor,
convert=make_tuple_converter (cursor))
def make_execute (fetch):
def execute (cursor, sql, **parameters):
cursor.execute (sql, parameters)
return fetch (cursor)
return execute
execute_none = make_execute (fetch_none)
execute_optional_value = make_execute (fetch_optional_value)
execute_required_value = make_execute (fetch_required_value)
execute_values = make_execute (fetch_values)
execute_optional_tuple = make_execute (fetch_optional_tuple)
execute_required_tuple = make_execute (fetch_required_tuple)
execute_tuples = make_execute (fetch_tuples)
def make_callproc (fetch):
def callproc (cursor, procname, *parameters):
cursor.callproc (procname, parameters)
return fetch (cursor)
return callproc
callproc_none = make_callproc (fetch_none)
callproc_optional_value = make_callproc (fetch_optional_value)
callproc_required_value = make_callproc (fetch_required_value)
callproc_values = make_callproc (fetch_values)
callproc_optional_tuple = make_callproc (fetch_optional_tuple)
callproc_required_tuple = make_callproc (fetch_required_tuple)
callproc_tuples = make_callproc (fetch_tuples)
def make_make_execute (fetch, sql):
def make_execute (cursor, **parameters):
cursor.execute (sql, parameters)
return fetch (cursor)
return make_execute
make_execute_none = partial (make_make_execute, fetch_none)
make_execute_optional_value = partial (make_make_execute, fetch_optional_value)
make_execute_required_value = partial (make_make_execute, fetch_required_value)
make_execute_values = partial (make_make_execute, fetch_values)
make_execute_optional_tuple = partial (make_make_execute, fetch_optional_tuple)
make_execute_required_tuple = partial (make_make_execute, fetch_required_tuple)
make_execute_tuples = partial (make_make_execute, fetch_tuples)
def make_make_callproc (fetch, procname):
def make_callproc (cursor, *parameters):
cursor.callproc (procname, parameters)
return fetch (cursor)
return make_callproc
make_callproc_none = partial (make_make_callproc, fetch_none)
make_callproc_optional_value = partial (make_make_callproc, fetch_optional_value)
make_callproc_required_value = partial (make_make_callproc, fetch_required_value)
make_callproc_values = partial (make_make_callproc, fetch_values)
make_callproc_optional_tuple = partial (make_make_callproc, fetch_optional_tuple)
make_callproc_required_tuple = partial (make_make_callproc, fetch_required_tuple)
make_callproc_tuples = partial (make_make_callproc, fetch_tuples)