# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2008, SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: db_utils.py 2014 2008-03-27 19:08:41Z pamor $
"""
Utilities for abstracting access to different databases via a consistent API.
"""
import string
from snaplogic.common.snap_exceptions import *
from snaplogic.common.snap_exceptions import *
def unique_values(iterable):
unique = []
seen = set()
for x in iterable:
if x not in seen:
unique.append(x)
seen.add(x)
return unique
def param_bind_vars(ordered_params, paramstyle):
"""
Create a dictionary mapping param_names to their bind variable string.
Creates a bind variable for each element in ordered_params to appropriately map to a value in an SQL
statement. The bind syntax used depends on the paramstyle value which should be obtained from the
DB-API2 compliant DB module.
The ordered_params argument should be an iteratable object that provides the parameters by name in the order they
occur within the query string. Although duplication of the parameters as they occur in the query string is not
required, since param_bind_container() requires that behavior, it's often easier to use the same value
here. Duplication will be ignored.
Example:
param_names = ['City', 'State', 'Zip', 'State']
param_bind_strings(param_names, 'qmark') => {'City': '?', 'State': '?', 'Zip': '?'}
param_bind_strings(param_names, 'numeric') => {'City': ':1', 'State': ':2', 'Zip': ':3'}
param_bind_strings(param_names, 'named') => {'City': ':City', 'State': ':State', 'Zip': ':Zip'}
param_bind_strings(param_names, 'format') => {'City': '%s', 'State': '%s', 'Zip': '%s'}
param_bind_strings(param_names, 'pyformat') => {'City': '%(City)s', 'State': '%(State)s', 'Zip': '%(Zip)s'}
@param ordered_params: A iterable container of parameter names to use for bind variables.
@type ordered_params: iterable container of strings
@param paramstyle: The parameter style used by the target database.
@type paramstyle: string
@return: Dictionary mapping parameter names to their bind variable string.
@rtype: dict
"""
unique_params = unique_values(ordered_params)
if paramstyle == 'qmark':
return dict([(param, '?') for param in unique_params])
elif paramstyle == 'numeric':
return dict([(param, ':' + str(i + 1)) for (i, param) in enumerate(unique_params)])
elif paramstyle == 'named':
return dict([(param, ':' + param) for param in unique_params])
elif paramstyle == 'format':
return dict([(param, '%s') for param in unique_params])
elif paramstyle == 'pyformat':
return dict([(param, '%(' + param + ')s') for param in unique_params])
else:
raise SnapValueError('Unexpected paramstyle: ' + paramstyle)
def param_bind_container(ordered_params, values, paramstyle):
"""
Create a proper container of values from the parameters.
Creates a container of the parameters for the given paramstyle. The ordered_params argument should be an iteratable
object that provides the parameters by name in the order they occur within the query string. Duplicate uses of the
same parameter must occur in the ordered_params iteration as well since certain paramstyle requirethe value
multiple times.
The values argument should provide a dictionary mapping the parameter names used in ordered_params to their value.
Example:
ordered_params = ['City', 'State', 'Zip', 'State']
values = {'City': 'San Mateo', 'State': 'CA', 'Zip': 94402}
param_bind_container(ordered_params, values, 'qmark') => ['San Mateo', 'CA', 94402, 'CA']
param_bind_container(ordered_params, values, 'numeric') => ['San Mateo', 'CA', 94402]
param_bind_container(ordered_params, values, 'named') => {'City': 'San Mateo', 'State': 'CA', 'Zip': 94402}
param_bind_container(ordered_params, values, 'format') => ['San Mateo', 'CA', 94402, 'CA']
param_bind_container(ordered_params, values, 'pyformat') => {'City': 'San Mateo', 'State': 'CA', 'Zip': 94402}
@param ordered_params: Iterable container of parameter names in the order they appear in the query string.
@type ordered_params: list of strings
@param values: Dictionary mapping parameter names to their value.
@type values: dict
@param paramstyle: The parameter style used by the target database.
@type paramstyle: string
@return: The parameter values in a container suitable for use in the execute() method of the database.
@rtype: dictionary or list
"""
if (paramstyle == 'qmark') or (paramstyle == 'format'):
return [values[param] for param in ordered_params]
elif (paramstyle == 'numeric') or (paramstyle == 'named') or (paramstyle == 'pyformat'):
unique_params = unique_values(ordered_params)
if paramstyle == 'numeric':
return [values[param] for param in unique_params]
else:
return values
else:
raise SnapValueError('Unexpected paramstyle: ' + paramstyle)
class SQLQueryBuilder(object):
def __init__(self, query, paramstyle=None):
self.paramstyle = paramstyle
self._template = string.Template(query)
self._identify_params()
query_template = property(lambda self: self._template.template)
def _identify_params(self):
# Use the regular expression that string.Template uses itself for identifying parameters, and build
# an in-order list of all found in the string.
params = []
pos = 0
query = self.query_template
param_re = self._template.pattern
while True:
# The regular expression defines (among others) two named patterns for parameters: named = $blah
# and braced = ${blah}. Only one should be defined per match.
m = param_re.search(query, pos)
if m is None:
break
elif m.group('named'):
params.append(m.group('named'))
pos = m.end('named') + 1
elif m.group('braced'):
params.append(m.group('braced'))
pos = m.end('braced') + 1
else:
break
self._ordered_params = params
def build(self, params, paramstyle=None):
if paramstyle is None:
paramstyle = self.paramstyle
if paramstyle is None:
raise SnapValueError("No paramstyle specified to build SQL query")
query = ""
param_vars = param_bind_vars(self._ordered_params, paramstyle)
query = self._template.safe_substitute(param_vars)
container = param_bind_container(self._ordered_params, params, paramstyle)
return (query, container)
|