# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2008 - 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: DBWrite.py 10330 2009-12-24 22:13:38Z grisha $
"""
DBWrite Module and Resource Definition.
This module contains the DBWrite Capability, ResDef, and the Component classes. This component provides
functionality of writing input view records to the specified database and table.
"""
import re
from decimal import Decimal
from copy import deepcopy
from urlparse import urlparse
import snaplogic.components as components
from snaplogic.cc import component_api
from snaplogic.cc.component_api import ComponentAPI
from snaplogic.components.DBComponent import DBComponent,COMMIT_METHOD_AUTO,COMMIT_METHOD_WHEN_DONE
from snaplogic.common.data_types import SnapString,SnapNumber,SnapDateTime
from snaplogic.common.snap_exceptions import SnapComponentError
from snaplogic.common import version_info
import snaplogic.cc.prop as prop
from snaplogic.snapi_base import keys
from snaplogic.components import DBUtils
from snaplogic import snapi
class DBWrite(DBComponent):
"""
This class implements the DB Write component, providing INSERT,
UPDATE or DELETE functionality.
"""
api_version = '1.0'
component_version = '1.2'
capabilities = {
ComponentAPI.CAPABILITY_INPUT_VIEW_LOWER_LIMIT : 1,
ComponentAPI.CAPABILITY_INPUT_VIEW_UPPER_LIMIT : 1,
ComponentAPI.CAPABILITY_OUTPUT_VIEW_LOWER_LIMIT : 0,
ComponentAPI.CAPABILITY_OUTPUT_VIEW_UPPER_LIMIT : 0
}
component_description = "Writes to DB"
component_label = "DB Writer"
component_doc_uri = "https://www.snaplogic.org/trac/wiki/Documentation/%s/ComponentRef/DBWrite" % \
version_info.doc_uri_version
# List of supported DML operations
dml_ops = ["insert", "update", "delete"]
# Integer representation of the sql commands.
_insertCommand = 1
_updateCommand = 2
_deleteCommand = 3
def create_resource_template(self):
"""
Create DBWrite resource definition template.
"""
self._create_common_db_props()
self.set_property_def('TableName',
prop.SimpleProp("Table Name",
SnapString,
"The name of the DB table being modified",
None,
True))
self.set_property_def('QueryCommand',
prop.SimpleProp("Query Command",
SnapString,
"The DML operation to perform on the specified DB table."
" (insert | update | delete)",
{"lov": self.dml_ops},
True))
self.set_property_def('WhereClause',
prop.SimpleProp("Where Clause",
SnapString,
"Qualifies the QueryCommand when its value is update or delete"))
self.set_property_value('WhereClause', '')
self._add_commit_method_prop()
def validate(self, err_obj):
"""
Validation of all required properties is done for us by the server.
Here we take care of any remaining special validation.
"""
self._common_db_validate(err_obj)
query_command = self.get_property_value('QueryCommand')
# Where: If a where clause is provided and it has an input view field reference, make sure it is valid.
where_clause = self.get_property_value("WhereClause")
if query_command == 'insert' and where_clause:
err_obj.get_property_err("WhereClause").set_message("WhereClause not allowed for 'insert' QueryCommand")
return
# 1678: Beware of null where clause!
if where_clause and not component_api.has_param(where_clause):
bind_cols = self.get_referenced_fields(where_clause, True)
input_views = self.list_input_view_names()
input_view = self.get_input_view_def(input_views[keys.SINGLE_VIEW])
input_view_fields = [ d[keys.FIELD_NAME] for d in input_view[keys.VIEW_FIELDS] ]
missing_fields = []
for bind_col in bind_cols:
if bind_col not in input_view_fields:
missing_fields.append("'%s'" % bind_col)
if missing_fields:
err_obj.get_property_err("WhereClause").set_message("Input field(s) %s not present in input view." % ','.join(missing_fields))
def _get_props_to_check_for_suggest(self):
"""
See L{DBComponent._get_props_to_check_for_suggest}.
"""
return ['TableName', 'WhereClause']
def _db_suggest_resource_values(self, err_obj, conn):
"""
Attempt to discover input view given connection info.
See L{DBComponent._db_suggest_resource_values}.
"""
table_name = self.get_property_value('TableName')
query_command = self.get_property_value('QueryCommand')
where_clause = self.get_property_value('WhereClause')
if not query_command:
if where_clause:
query_command = 'update'
else:
query_command = 'insert'
self.set_property_value('QueryCommand', query_command)
else:
query_command = query_command.lower()
if query_command == 'insert' and where_clause:
err_obj.get_property_err("WhereClause").set_message("WhereClause not allowed for 'insert' command")
return
view_metadata = conn.get_snap_view_metadata(table_name)
fields = view_metadata['fields']
view_name = 'Input'
in_views = self.list_input_view_names()
if in_views:
view_name = in_views[0]
self.remove_input_view_def(in_views[0])
self.add_record_input_view_def(view_name, fields, table_name)
pkey = view_metadata['primary_key']
if pkey and query_command != 'insert' and not where_clause:
pkey.sort()
where_clause = []
for col in pkey:
where_clause.append("%s = ${%s}" % (col, col))
where_clause = ' AND '.join(where_clause)
self.set_property_value('WhereClause', where_clause)
def upgrade_1_0_to_1_1(self):
self._upgrade_field_refs("WhereClause")
self._upgrade_to_use_refs()
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
def execute(self, input_views, output_views):
try:
input_view = input_views.values()[keys.SINGLE_VIEW]
except IndexError:
raise SnapComponentError("No input view connected.")
connect_resdef = self.get_referenced_resdef("DBConnect")
if not connect_resdef:
connect_uri = self.get_property_value("DBConnect")
connect_resdef = self.get_local_resource_object(connect_uri)
self._db = DBUtils.get_connection_from_resdef(connect_resdef)
self._cursor = self._db.cursor()
self._table = self.get_property_value('TableName')
# 1678: Beware of null where clause!
self._where = self.get_property_value('WhereClause')
if self._where:
self._where = self._where.strip()
else:
self._where = ''
self._query = self.get_property_value('QueryCommand').lower()
self._method = self.get_property_value('CommitMethod').lower()
# Setup flag value for faster comparison later when processing each record.
# Flags for the commit methods.
self._autoCommit = False
self._commitWhenDone = False
if self._method == COMMIT_METHOD_AUTO:
self._autoCommit = True
elif self._method == COMMIT_METHOD_WHEN_DONE:
self._commitWhenDone = True
else:
raise SnapComponentError("Unsupported commit method %s" % self._method)
# Initializes that part in query statement that is constant for all record operations.
# It is assumed that the query command has been validated.
# Also set integer for faster comparison purpose.
# self._query is not used for now outside of init(), but kept in as instance variable for debugging purpose.
bind_vars = []
if self._query == 'insert':
self._sqlcommand = DBWrite._insertCommand
bind_vars = self._db.bindVariableList(input_view.field_names)
sqlstmtbase = 'INSERT INTO %s (%s) VALUES (%s)' % (self._table,
', '.join(input_view.field_names),
', '.join(bind_vars))
elif self._query == 'update':
self._sqlcommand = DBWrite._updateCommand
bind_vars = self._db.bindVariableList(input_view.field_names)
assignments = ['%s = %s' % (input_view.field_names[i], bind_vars[i])
for i in range(0, len(input_view.field_names))]
sqlstmtbase = 'UPDATE %s SET %s' % (self._table, ', '.join(assignments))
elif self._query == 'delete':
self._sqlcommand = DBWrite._deleteCommand
sqlstmtbase = 'DELETE FROM ' + self._table
if self._where:
where_fields_to_replace = self.get_referenced_fields(self._where)
where_bind_vars = self._db.bindVariableList(where_fields_to_replace)
replacements = {}
for i in range(len(where_fields_to_replace)):
replacements[where_fields_to_replace[i]] = where_bind_vars[i]
where_clause = self.replace_referenced_fields(self._where, replacements)
sqlstmtbase += " WHERE %s" % where_clause
try:
record = input_view.read_record()
while record is not None:
value_cont = self._db.bindValueContainer(record)
if self._where:
if isinstance(value_cont, list):
# In case of dictionary-based value containers,
# we don't need this extra step.
where_value_cont = [record[field_name] for field_name in where_fields_to_replace]
if hasattr(self._db, 'fix_bound_values'):
where_value_cont = self._db.fix_bound_values(where_value_cont)
if self._query == 'delete':
value_cont = where_value_cont
else:
value_cont.extend(where_value_cont)
if self._query == 'delete' and not self._where:
self._cursor.execute(sqlstmtbase)
else:
self._cursor.execute(sqlstmtbase, value_cont)
# Commit the record is configured so.
if self._autoCommit:
self._db.commit()
record = input_view.read_record()
if self._commitWhenDone:
self._db.commit()
finally:
if hasattr(self, '_cursor'):
self._cursor.close()
if hasattr(self, '_db'):
self._db.close()
|