# $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: DBConnection.py 10183 2009-12-16 23:58:55Z grisha $
"""
Contains superclass for DB Connection components.
"""
from snaplogic.cc.component_api import ComponentAPI
from snaplogic.cc import component_api
from snaplogic.components import DBUtils
class DBConnection(ComponentAPI):
"""
This class is what would be in Java an "abstract" class, containing
functionality that is common for Connection components, but not a
Component itself.
"""
capabilities = {
ComponentAPI.CAPABILITY_INPUT_VIEW_LOWER_LIMIT : 0,
ComponentAPI.CAPABILITY_INPUT_VIEW_UPPER_LIMIT : 0,
ComponentAPI.CAPABILITY_OUTPUT_VIEW_LOWER_LIMIT : 0,
ComponentAPI.CAPABILITY_OUTPUT_VIEW_UPPER_LIMIT : 0,
}
def validate(self, err_obj):
"""
Attempt to connect to the database with current properties.
If current properties are not parameterized and connection
cannot be made, then an error is set. See
L{ComponentAPI.validate()} for more information.
"""
conn = None
resdef = self._resdef
for prop_name in resdef.list_property_names():
prop_val = self.get_property_value(prop_name)
if component_api.has_param(prop_val):
# Can't do anything...
return
try:
conn = DBUtils.get_connection_from_resdef(self._resdef)
except Exception, e:
self.elog(e)
err_obj.set_message("Error getting connection: %s" % e)
finally:
if conn:
try:
conn.close()
except:
pass
def execute(self, input_views, output_views):
"""A noop."""
pass
def _set_cats(self):
"""Set the category to "connection.db.<DB-type>" """
cat = "connection.db."
cls_name = self.__class__.__name__
cat += cls_name[len('Connection'):]
self.set_categories([cat],False)
|