# $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: DBUpsert.py 10330 2009-12-24 22:13:38Z grisha $
"""
DB Upsert component - update DB rows if values for certain columns
exist, otherwise insert new row.
"""
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.common.data_types import SnapString,SnapNumber,SnapDateTime
from snaplogic.common.snap_exceptions import SnapComponentError
from snaplogic.components.DBComponent import DBComponent,COMMIT_METHOD_AUTO,COMMIT_METHOD_WHEN_DONE
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 DBUpsert(DBComponent):
"""
DB Upsert component - update DB rows if values for certain columns
exist, otherwise insert new row.
"""
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 = "Upserts (updates or inserts) rows to Database"
component_label = "DB Upsert"
component_doc_uri = "https://www.snaplogic.org/trac/wiki/Documentation/%s/ComponentRef/DBUpsert" % \
version_info.doc_uri_version
def create_resource_template(self):
"""
Create DBUpsert resource definition template.
"""
self._create_common_db_props()
key = prop.SimpleProp("Key",
SnapString,
"""Input field (corresponding to database column name) which is to serve as one of the keys based on which we decide whether to update or insert.""",
{'lov' : [keys.CONSTRAINT_LOV_INPUT_FIELD] },
True)
keys_prop = prop.ListProp("Keys",
key,
"""
Input fields (corresponding to database column names) which serve as keys based on which we decide whether to update or insert.
""",
1,
required=True)
self.set_property_def('Keys', keys_prop)
self.set_property_def('TableName',
prop.SimpleProp("Table Name",
SnapString,
"The name of the DB table being modified",
None,
True))
self._add_commit_method_prop()
def _get_props_to_check_for_suggest(self):
"""
See L{DBComponent._get_props_to_check_for_suggest}.
"""
return ['TableName']
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')
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:
self.set_property_value('Keys', pkey)
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.
"""
# DBConnect: Just check that this looks like a URI that begins with http:// or /
self._common_db_validate(err_obj)
keys_prop = self.get_property_value('Keys')
repeated = []
for key in keys_prop:
if keys_prop.count(key) > 1 and key not in repeated:
repeated.append(key)
if repeated:
err = err_obj.get_property_err('Keys')
err.set_message("The following keys appear more than once: %s" % ', '.join(repeated))
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._table = self.get_property_value('TableName')
self._keys = self.get_property_value('Keys')
self._method = self.get_property_value('CommitMethod').lower()
# 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)
# It is better to validate the symbols against input view fields here, instead of waiting until
# processing the first record to raise an exception.
# But how to get access to the input view field names?
table_metadata = self._db.get_snap_view_metadata(self._table)
try:
record = input_view.read_record()
while record is not None:
self._db.upsert(self._table, record, self._keys, table_metadata)
record = input_view.read_record()
if self._autoCommit:
self._db.commit()
if self._commitWhenDone:
self._db.commit()
finally:
if hasattr(self, '_db'):
self._db.close()
def upgrade_1_0_to_1_1(self):
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
|