# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2009, 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: string_math.py 10330 2009-12-24 22:13:38Z grisha $
"""
Base class for StringExpressions and MathExpressions components.
This module does all the actual work.
"""
import os
import re
from sqlite3 import dbapi2
from decimal import Decimal
from sets import Set
import snaplogic.components as components
from snaplogic.cc import component_api
from snaplogic.cc.component_api import ComponentAPI
from snaplogic.cc import prop
from snaplogic.common import snap_log
from snaplogic.snapi_base import resdef,keys
from snaplogic.common.data_types import SnapNumber,SnapString
from snaplogic.common.snap_exceptions import *
FUNCTION = "Expression"
OUTPUT_FIELD_NAME = "Output view field name"
MATH_SPEC = "Expression specification"
MATH_SPECS = "Expressions"
class StringMath(ComponentAPI):
# NB: absent "api_version = '1.0'" line makes this base class not get picked as a component.
capabilities = {
ComponentAPI.CAPABILITY_INPUT_VIEW_LOWER_LIMIT : 1,
ComponentAPI.CAPABILITY_INPUT_VIEW_UPPER_LIMIT : 1,
ComponentAPI.CAPABILITY_OUTPUT_VIEW_LOWER_LIMIT : 1,
ComponentAPI.CAPABILITY_OUTPUT_VIEW_UPPER_LIMIT : 1,
ComponentAPI.CAPABILITY_ALLOW_PASS_THROUGH : True
}
# We don't support datetime type in this component. Use DateMath.
supported_datatypes = [SnapString, SnapNumber]
# For finding input field name references in the expressions in the form ${Field001}
_subst_re = re.compile('\$[{]([\w]+)[}]')
def _create_resource_template(self):
"""
Create StringMath resource template.
"""
func = prop.SimpleProp(FUNCTION,
"string",
"Defines an SQL expression that is to be evaluated",
None,
True)
output_field_name = prop.SimpleProp(OUTPUT_FIELD_NAME,
"string",
"What output field the result corresponds to",
{'lov': [ keys.CONSTRAINT_LOV_OUTPUT_FIELD] },
True)
math_spec = prop.DictProp(MATH_SPEC,
None,
"Expression specification dictionary",
2,
2,
True,
True)
math_spec[FUNCTION] = func
math_spec[OUTPUT_FIELD_NAME] = output_field_name
math_specs = prop.ListProp(MATH_SPECS,
math_spec,
"",
1,
resdef.UNLIMITED_ENTRIES,
True)
self.set_property_def(MATH_SPECS, math_specs)
def _validate(self, err_obj):
"""
Component-specific correctness validation logic.
@param err_obj: Object for error reporting
@type err_obj: L{SimplePropErr} or L{ListPropErr} or L{DictPropErr}
"""
output_views = self.list_output_view_names()
output_view_name = output_views[keys.SINGLE_VIEW]
output_view = self.get_output_view_def(output_view_name)
field_names = [ d[keys.FIELD_NAME] for d in output_view[keys.VIEW_FIELDS] ]
field_types = [ d[keys.FIELD_TYPE] for d in output_view[keys.VIEW_FIELDS] ]
input_views = self.list_input_view_names()
input_view_name = input_views[keys.SINGLE_VIEW]
input_view = self.get_input_view_def(input_view_name)
input_field_names = [ d[keys.FIELD_NAME] for d in input_view[keys.VIEW_FIELDS] ]
input_field_types = [ d[keys.FIELD_TYPE] for d in input_view[keys.VIEW_FIELDS] ]
math_specs = self.get_property_value(MATH_SPECS)
# NB: We use the list of expression_output_field_names later in validating the output view fields.
expression_output_field_names = []
for i, spec in enumerate(math_specs):
output_field_name = spec[OUTPUT_FIELD_NAME]
expression_output_field_names.append(output_field_name)
func = spec[FUNCTION]
# Check 1: For each field, check the datatype.
# If we don't support this datatype add an error.
index = field_names.index(output_field_name)
field_type = field_types[index]
if field_type not in self.supported_datatypes:
err_obj.get_output_view_err()[output_view_name][keys.VIEW_FIELDS][i].set_message(
"Output field '%s' datatype '%s' is not supported. Must be one of: %s" %
(output_field_name, field_type, str(self.supported_datatypes)))
# Check 2: input view field references of the form ${field001} in expressions
# match input view field names.
for input_field in self._subst_re.findall(func):
if input_field not in input_field_names:
err_obj.get_property_err(MATH_SPECS)[i][FUNCTION].set_message(
"Input field name '%s' not present in input view." % input_field)
# Check 4: All output fields must have a corresponding expression,
# except for output fields named the same as input fields.
# If output field name equals input field name we transfer the value unchanged,
# unless there is an expression defined that sends output to this field.
for i, field in enumerate(field_names):
if field in expression_output_field_names:
# There is an expression defined on this field.
# There is nothing else to validate, move on to the next field.
pass
elif field in input_field_names:
# If output field name is the same as input field name.
# the field value will be transferred unchanged, so validate the type
input_field_type = input_field_types[input_field_names.index(field)]
output_field_type = field_types[field_names.index(field)]
if input_field_type != output_field_type:
err_obj.get_output_view_err()[output_view_name][keys.VIEW_FIELDS][i].set_message(
"Output view '%s' field '%s' type '%s' does not match corresponding input view '%s' field '%s' type '%s'." %
(output_view_name, field, output_field_type, input_view_name, field, input_field_type))
else:
# There is no expression defined for this output field, neither
# there is an input field with the same name: log an error.
err_obj.get_output_view_err()[output_view_name][keys.VIEW_FIELDS][i].set_message(
"Output view field '%s' does not have a corresponding expression." % field)
def _cleanup(self):
"""
Clean up resources.
"""
if self._cur:
try:
self._cur.close()
except:
pass
if self._con:
try:
self._con.close()
except:
pass
def _execute(self, input_views, output_views):
try:
self._inp_view = input_views[self.list_input_view_names()[keys.SINGLE_VIEW]]
except IndexError:
raise SnapComponentError("No output view connected.")
try:
self._out_view = output_views.values()[keys.SINGLE_VIEW]
except IndexError:
raise SnapComponentError("No output view connected.")
try:
self._exec()
finally:
self._cleanup()
def _exec(self):
# Using sqlite in this way provides us a safe "sandbox" for executing
# sqlite statements in-memory only during the lifetime of this component.
self._con = sqlite.connect( ":memory:" )
self._cur = self._con.cursor()
input_field_names = self._inp_view.field_names
input_field_types = []
for input_field in self._inp_view.fields:
input_field_types.append(input_field[keys.FIELD_TYPE])
out_field_names = list(self._out_view.field_names)
output_field_types = [output_field[keys.FIELD_TYPE] for output_field in self._out_view.fields]
# Make a list of common fields: fields with same names in the input and output views
common_fields = Set(input_field_names) & Set(out_field_names)
math_specs = self.get_property_value(MATH_SPECS)
self.log(snap_log.LEVEL_DEBUG, "StringMath math_specs: %s" % math_specs)
# Make a map of expressions corresponding to output field names
output_name_to_function = {}
for spec in math_specs:
output_name_to_function[spec[OUTPUT_FIELD_NAME]] = spec[FUNCTION]
self.log(snap_log.LEVEL_DEBUG, "StringMath: output_name_to_function: %s" % output_name_to_function)
out_rec = self._out_view.create_record()
first = True
# vals_names stores an ordered list of input field names as they are encountered (as
# parameters) in the expressions. If the same input field is used in multiple properties,
# it will be repeated in this list.
vals_names = []
# stmt collects all expressions into a single comma-separated sqlite statement
stmt = "SELECT "
used_output_fields = []
for f in out_field_names:
if output_name_to_function.has_key(f):
used_output_fields.append(f)
func = output_name_to_function[f]
# Extract all tokens wrapped inside '${token}'
flds = re.findall('\${(\w+)}', func)
for fld in flds:
# Check if this is a field name in inputview
if fld in input_field_names:
# Substitute field name with '?', which will later (at the time of sqlite
# statement execution be substituted with its value, using vals_names for lookup.
func = re.sub('\${' + fld + '}', '?', func)
vals_names.append(fld)
else:
raise SnapComponentError("Unexpected field %s that is not present in the input view encountered in expression %s" % fld, func)
if not first:
stmt += ", "
else:
first = False
stmt += "%s" % func
self.log(snap_log.LEVEL_DEBUG, "StringMath: sqlite statement: %s" % stmt)
# Collect the types of used output fields into an array
used_output_field_types = []
for f in used_output_fields:
index = out_field_names.index(f)
field = self._out_view.fields[index]
used_output_field_types.append(field[keys.FIELD_TYPE])
while True:
record = self._inp_view.read_record()
if record is None:
break
# vals list stores actual values for input fields which names were
# collected (above) from the expressions properties.
vals = []
for field_name in vals_names:
val = record[field_name]
if isinstance(val, Decimal):
vals.append(str(val))
else:
vals.append(val)
self.log(snap_log.LEVEL_DEBUG, "StringMath: input values list: %s" % vals)
self._cur.execute(stmt, vals)
# Read the results of sqlite statement execution into records
for row in self._cur:
self.log(snap_log.LEVEL_DEBUG, "StringMath: result row: %s" % str(row))
i = 0
# Create a record and transfer passthrough fields from input record,
# as well as fields matched by name
out_rec = self._out_view.create_record()
out_rec.transfer_pass_through_fields(record)
out_rec.transfer_matching_fields(record, common_fields)
for field in used_output_fields:
value = row[i]
if value is None:
out_rec[field] = None
elif used_output_field_types[i] == SnapNumber:
if isinstance(value, float):
out_rec[field] = Decimal(str(value))
else:
out_rec[field] = Decimal(value)
elif used_output_field_types[i] == SnapString:
out_rec[field] = unicode(value)
i += 1
self._out_view.write_record(out_rec)
self._out_view.completed()
def upgrade_1_0_to_1_1(self):
"""
Add source constraint to Field property
"""
# Save the property value.
# We need to recreate the property, which resets the value
property_value = self.get_property_value(MATH_SPECS)
func = prop.SimpleProp(FUNCTION,
"string",
"Defines an SQL expression that is to be evaluated",
None,
True)
output_field_name = prop.SimpleProp(OUTPUT_FIELD_NAME,
"string",
"What output field the result corresponds to",
{'lov': [ keys.CONSTRAINT_LOV_OUTPUT_FIELD] },
True)
math_spec = prop.DictProp(MATH_SPEC,
None,
"Expression specification dictionary",
2,
2,
True,
True)
math_spec[FUNCTION] = func
math_spec[OUTPUT_FIELD_NAME] = output_field_name
math_specs = prop.ListProp(MATH_SPECS,
math_spec,
"",
1,
resdef.UNLIMITED_ENTRIES,
True)
self.set_property_def(MATH_SPECS, math_specs)
# Restore the value
self.set_property_value(MATH_SPECS, property_value)
def upgrade_1_1_to_1_2(self):
"""
No-op upgrade only to change component doc URI during the upgrade
which will be by cc_info before calling this method.
"""
pass
|