Possibly you can see how my implementation is off. Any advice you can offer would be greatly appreciated.
"""
AyxPlugin (required) has-a IncomingInterface (optional).
Although defining IncomingInterface is optional, the interface methods are needed if an upstream tool exists.
"""
import ibm_db
import AlteryxPythonSDK as Sdk
import xml.etree.ElementTree as Et
from tkinter import messagebox
# ***************************************************************************************************************************************************
# Static method that I am using in one of the Classes
# ***************************************************************************************************************************************************
def is_empty(any_structure):
if any_structure:
return False
else:
return True
class AyxPlugin:
"""
Implements the plugin interface methods, to be utilized by the Alteryx engine to communicate with a plugin.
Prefixed with "pi", the Alteryx engine will expect the below five interface methods to be defined.
"""
def __init__(self, n_tool_id: int, alteryx_engine: object, output_anchor_mgr: object):
"""
Constructor is called whenever the Alteryx engine wants to instantiate an instance of this plugin.
:param n_tool_id: The assigned unique identification for a tool instance.
:param alteryx_engine: Provides an interface into the Alteryx engine.
:param output_anchor_mgr: A helper that wraps the outgoing connections for a plugin.
"""
# Default properties
self.n_tool_id = n_tool_id
self.alteryx_engine = alteryx_engine
self.output_anchor_mgr = output_anchor_mgr
# Custom properties
self.user = ''
self.password = ''
self.database = ''
self.sql = ''
self.str_file_path = ''
self.error_list = ''
self.user_input = None
self.sql_input = None
def pi_init(self, str_xml: str):
"""
Handles building out the sort info, to pass into pre_sort() later on, from the user configuration.
Called when the Alteryx engine is ready to provide the tool configuration from the GUI.
:param str_xml: The raw XML from the GUI.
"""
self.user = Et.fromstring(str_xml).find('User').text
if self.user is None:
self.user = ''
self.password = Et.fromstring(str_xml).find('Password').text
if self.password is None:
self.password = ''
if self.password != '':
self.password = self.alteryx_engine.decrypt_password(self.password, 0)
self.database = Et.fromstring(str_xml).find('Database').text
if self.database is None:
self.database = ''
self.sql = Et.fromstring(str_xml).find('SQL').text
if self.sql is None:
self.sql = ''
self.output_anchor = self.output_anchor_mgr.get_output_anchor('Output') # Getting the output anchor from the XML file.
def pi_add_incoming_connection(self, str_type: str, str_name: str) -> object:
"""
The IncomingInterface objects are instantiated here, one object per incoming connection, also pre_sort() is called here.
Called when the Alteryx engine is attempting to add an incoming data connection.
:param str_type: The name of the input connection anchor, defined in the Config.xml file.
:param str_name: The name of the wire, defined by the workflow author.
:return: The IncomingInterface object(s).
"""
if str_type == 'SQL':
self.sql_input = SQLInterface(self)
messagebox.showinfo('pi_add', 'SQL')
return self.sql_input
elif str_type == 'Input':
self.user_input = UserInterface(self)
messagebox.showinfo('pi_add', 'User')
return self.user_input
def pi_add_outgoing_connection(self, str_name: str) -> bool:
"""
Called when the Alteryx engine is attempting to add an outgoing data connection.
:param str_name: The name of the output connection anchor, defined in the Config.xml file.
:return: True signifies that the connection is accepted.
"""
return True
def pi_push_all_records(self, n_record_limit: int) -> bool:
"""
Called when a tool has no incoming data connection.
:param n_record_limit: Set it to <0 for no limit, 0 for no records, and >0 to specify the number of records.
:return: True for success, False for failure.
"""
return False
def pi_close(self, b_has_errors: bool):
"""
Called after all records have been processed..
:param b_has_errors: Set to true to not do the final processing.
"""
#First thing first. If this is a "config" run, we are going to bail early, i.e. not run the code
# if self.alteryx_engine.get_init_var(self.n_tool_id, 'UpdateOnly') == 'True':
# headers = ['Results']
# types = ['string']
# record_info_out = self.build_record_info_out(headers, types) # Building out the outgoing record layout.
# self.output_anchor.init(record_info_out) # Lets the downstream tools know of the outgoing record metadata.
# return False
messagebox.showinfo('pi_close', self.sql_input.get_sql()[10:])
# Check for Incoming connections and set the appropriate variables
if type(self.sql_input) is SQLInterface:
self.sql = self.sql_input.get_sql()
if type(self.user_input) is UserInterface:
self.user = self.user_input.get_user()
self.password = self.user_input.get_password()
self.database = self.user_input.get_database()
# messagebox.showinfo('SQL', self.sql)
# If there is no incoming interface (i.e. user id and password)
error_sql = self.msg_sql(self.sql)
error_database = self.msg_database(self.database)
error_racf_id = self.msg_racfid(self.user)
error_password = self.msg_password(self.password)
if error_sql != '':
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.info, self.xmsg(error_sql))
return False
if error_database != '':
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.info, self.xmsg(error_database))
return False
elif error_racf_id != '':
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.error, self.xmsg(error_racf_id))
return False
elif error_password != '':
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.error, self.xmsg(error_password))
return False
query = IBMCall("3700", "TCPIP", 'DB' + str(self.database)[:2], str(self.user) + 'B', str(self.password), self.sql.upper().replace('&DATABASE.', self.database))
result = query.get_data()[1:]
headers = query.get_headers()
types = query.get_types()
# messagebox.showinfo('SQL', result)
# We won't stop the code if the IBM call broke, but we will pop a message
self.error_list = query.get_errorlist()
if self.error_list != '':
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.info, self.error_list)
total_records = len(result)
if total_records == 0:
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.info, 'Query returned no results.')
return False
record_info_out = self.build_record_info_out(headers, types) # Building out the outgoing record layout.
self.output_anchor.init(record_info_out) # Lets the downstream tools know of the outgoing record metadata.
# Create Output Object
record_creator = record_info_out.construct_record_creator() # Creating a new record_creator for the new data.
# Loop Results
for record in enumerate(result):
for field in enumerate(record[1]):
if types[field[0]] == 'int':
record_info_out[field[0]].set_from_int64(record_creator, int(field[1]))
if types[field[0]] == 'string':
record_info_out[field[0]].set_from_string(record_creator, str(field[1]))
if types[field[0]] == 'date':
record_info_out[field[0]].set_from_string(record_creator, str(field[1]))
if types[field[0]] == 'timestamp':
record_info_out[field[0]].set_from_string(record_creator, str(field[1]))
if types[field[0]] == 'decimal':
record_info_out[field[0]].set_from_double(record_creator, float(field[1]))
# Asking for a record to push downstream
out_record = record_creator.finalize_record()
self.output_anchor.push_record(out_record, False) # False: completed connections will automatically close.
# Not the best way to let the downstream tool know of this tool's progress, normally one would use a timer.
self.output_anchor.update_progress(record[0] / float(total_records))
record_creator.reset() # Resets the variable length data to 0 bytes (default) to prevent unexpected results.
# self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.info, self.xmsg(str(total_records)) + ' records were read')
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.info, str(total_records) + ' rows completed')
self.output_anchor.close() # Close outgoing connections.
# self.output_anchor.assert_close() # Checks whether connections were properly closed.
def display_error_message(self, msg_string: str):
"""
A non-interface helper function, responsible for outputting error messages.
:param msg_string: The error message string.
"""
self.alteryx_engine.output_message(self.n_tool_id, Sdk.EngineMessageType.error, self.xmsg(msg_string))
def xmsg(self, msg_string: str):
"""
A non-interface, non-operational placeholder for the eventual localization of predefined user-facing strings.
:param msg_string: The user-facing string.
:return: msg_string
"""
return msg_string
def build_record_info_out(self, field_headers, field_types):
"""
A non-interface helper for pi_push_all_records() responsible for creating the outgoing record layout.
:param file_reader: The name for csv file reader.
:return: The outgoing record layout, otherwise nothing.
"""
record_info_out = Sdk.RecordInfo(self.alteryx_engine) # A fresh record info object for outgoing records.
for field in enumerate(field_headers):
if field_types[field[0]] == 'int':
record_info_out.add_field(field[1], Sdk.FieldType.int64, 256, 0, '', '')
elif field_types[field[0]] == 'string':
record_info_out.add_field(field[1], Sdk.FieldType.v_wstring, 256, 0, '', '')
elif field_types[field[0]] == 'date':
record_info_out.add_field(field[1], Sdk.FieldType.date, 256, 0, '', '')
elif field_types[field[0]] == 'decimal':
record_info_out.add_field(field[1], Sdk.FieldType.double, 256, 16, '', '')
elif field_types[field[0]] == 'timestamp':
record_info_out.add_field(field[1], Sdk.FieldType.datetime, 256, 0, '', '')
return record_info_out
@staticmethod
def msg_sql(sql: str):
"""
A non-interface, helper function that handles validating the file path input.
:param mainframe: The mainframe file to get via FTP
:return: The chosen message string.
"""
msg_sql = ''
if len(sql) == 0:
msg_sql = 'Enter a SQL query'
return msg_sql
@staticmethod
def msg_database(database: str):
"""
A non-interface, helper function that handles validating the file path input.
:param mainframe: The mainframe file to get via FTP
:return: The chosen message string.
"""
msg_database = ''
if len(database) == 0:
msg_database = 'Enter a database name'
return msg_database
@staticmethod
def msg_racfid(racfid: str):
"""
A non-interface, helper function that handles validating the file path input.
:param racfid: The user's racfid
:return: The chosen message string.
"""
msg_racfid = ''
if len(racfid) == 0:
msg_racfid = 'Enter a RACF ID'
elif len(racfid) > 6:
msg_racfid = 'The maximum length for RACF ID is 6 (use non-B-id)'
elif racfid[0] != '@':
msg_racfid = 'Invalid RACF ID: must begin with @'
return msg_racfid
@staticmethod
def msg_password(password: str):
"""
A non-interface, helper function that handles validating the file path input.
:param password: The user's password
:return: The chosen message string.
"""
msg_password = ''
if len(password) == 0:
msg_password = 'Enter a Password'
return msg_password
class UserInterface:
"""
This optional class is returned by pi_add_incoming_connection, and it implements the incoming interface methods, to
be utilized by the Alteryx engine to communicate with a plugin when processing an incoming connection.
Prefixed with "ii", the Alteryx engine will expect the below four interface methods to be defined.
"""
def __init__(self, parent: object):
"""
Constructor for IncomingInterface.
:param parent: AyxPlugin
"""
# Default properties
self.parent = parent
# Custom members
self.record_info_in = None
self.field_lists = []
self.counter = 0
self.user = ''
self.password = ''
self.database = ''
self.sql = ''
self.error_list = ''
self.i_ran = False
def ii_init(self, record_info_in: object) -> bool:
"""
Handles the storage of the incoming metadata for later use.
Called to report changes of the incoming connection's record metadata to the Alteryx engine.
:param record_info_in: A RecordInfo object for the incoming connection's fields.
:return: True for success, otherwise False.
"""
self.record_info_in = record_info_in # For later reference.
# Storing the field names to use when writing data out.
for field in range(record_info_in.num_fields):
self.field_lists.append([record_info_in[field].name])
return True
def ii_push_record(self, in_record: object) -> bool:
"""
Responsible for writing the data to csv in chunks.
Called when an input record is being sent to the plugin.
:param in_record: The data for the incoming record.
:return: False if file path string is invalid, otherwise True.
"""
if self.i_ran == False:
messagebox.showinfo("ii_push", "User")
self.i_ran = True
#First thing first. If this is a "config" run, we are going to bail early, i.e. not run the code
if self.parent.alteryx_engine.get_init_var(self.parent.n_tool_id, 'UpdateOnly') == 'True':
return False
for field in range(self.record_info_in.num_fields):
if self.record_info_in[field].name == 'UserName':
self.user = self.record_info_in[field].get_as_string(in_record)
elif self.record_info_in[field].name == 'Password':
self.password = self.record_info_in[field].get_as_string(in_record)
if self.password != '':
self.password = self.parent.alteryx_engine.decrypt_password(self.password, 0)
elif self.record_info_in[field].name == 'Database':
self.database = self.record_info_in[field].get_as_string(in_record)
return True
def ii_update_progress(self, d_percent: float):
"""
Called by the upstream tool to report what percentage of records have been pushed.
:param d_percent: Value between 0.0 and 1.0.
"""
self.parent.alteryx_engine.output_tool_progress(self.parent.n_tool_id, d_percent) # Inform the Alteryx engine of the tool's progress
def ii_close(self):
"""
Handles writing out any residual data out.
Called when the incoming connection has finished passing all of its records.
"""
messagebox.showinfo('ii_close', 'User Class')
# GETTERS AND SETTERS
def get_user(self):
return self.user
def get_password(self):
return self.password
def get_database(self):
return self.database
class SQLInterface:
"""
This optional class is returned by pi_add_incoming_connection, and it implements the incoming interface methods, to
be utilized by the Alteryx engine to communicate with a plugin when processing an incoming connection.
Prefixed with "ii", the Alteryx engine will expect the below four interface methods to be defined.
"""
def __init__(self, parent: object):
"""
Constructor for IncomingInterface.
:param parent: AyxPlugin
"""
# Default properties
self.parent = parent
# Custom members
self.record_info_in = None
self.field_lists = []
self.counter = 0
self.sql = ''
self.error_list = ''
self.add_num = 0
self.i_ran = False
def ii_init(self, record_info_in: object) -> bool:
"""
Handles the storage of the incoming metadata for later use.
Called to report changes of the incoming connection's record metadata to the Alteryx engine.
:param record_info_in: A RecordInfo object for the incoming connection's fields.
:return: True for success, otherwise False.
"""
self.record_info_in = record_info_in # For later reference.
# Storing the field names to use when writing data out.
for field in range(record_info_in.num_fields):
self.field_lists.append([record_info_in[field].name])
return True
def ii_push_record(self, in_record: object) -> bool:
"""
Responsible for writing the data to csv in chunks.
Called when an input record is being sent to the plugin.
:param in_record: The data for the incoming record.
:return: False if file path string is invalid, otherwise True.
"""
self.add_num += 1
if self.i_ran == False:
messagebox.showinfo("ii_push", "SQL")
self.i_ran = True
#First thing first. If this is a "config" run, we are going to bail early, i.e. not run the code
if self.parent.alteryx_engine.get_init_var(self.parent.n_tool_id, 'UpdateOnly') == 'True':
return False
for field in range(self.record_info_in.num_fields):
line = self.record_info_in[field].get_as_string(in_record).strip()
if line[:1] != '#' and line.upper() != 'WITH UR' and line is not None and line != '':
self.sql += line + '\n'
return True
def ii_update_progress(self, d_percent: float):
"""
Called by the upstream tool to report what percentage of records have been pushed.
:param d_percent: Value between 0.0 and 1.0.
"""
self.parent.alteryx_engine.output_tool_progress(self.parent.n_tool_id, d_percent) # Inform the Alteryx engine of the tool's progress
def ii_close(self):
"""
Handles writing out any residual data out.
Called when the incoming connection has finished passing all of its records.
"""
messagebox.showinfo('ii_close', "SQL Object")
# GETTERS AND SETTERS
def get_sql(self):
return self.sql
class IBMCall:
def __init__(self, port, protocol, database, user, password, sql, driver="{IBM DB2 ODBC DRIVER}", host="XXXX"):
# Connection Items
self.driver = driver
self.host = host
self.port = port
self.protocol = protocol
self.database = database
self.user = user
self.password = password
self.sql = sql
self.error_list = ''
# Looping items initialize
self.temp = []
self.headers = []
first_row = []
self.data_type = []
# This is the return ITEM for this Object..It is the noun.
self.q_list = []
# Connection String
self.conn_string = 'DRIVER=' + self.driver + ';HOSTNAME=' + self.host + ';PORT=' + self.port + ';PROTOCOL=' + self.protocol + ';DATABASE=' + self.database + ';UID=' + self.user + ';PWD=' + self.password
# Connection
self.cnxn = self.make_connection()
self.query_stmt = self.run_sql()
if self.error_list == '':
try:
# The data is coming in not stripped. I have to strip it all out
self.tup = ibm_db.fetch_both(self.query_stmt)
i = -1
for item, val in self.tup.items():
i += 1
if (i / 2) == int(i / 2) or i == 0:
self.headers.append(str(item).strip())
first_row.append(str(val).strip())
self.data_type.append(ibm_db.field_type(self.query_stmt, item))
self.q_list.append(self.headers)
self.q_list.append(first_row)
self.tup = ibm_db.fetch_tuple(self.query_stmt)
# Read it in and strip out all the garbage
while not is_empty(self.tup):
for self.each in self.tup:
self.temp.append(str(self.each).strip())
self.q_list.append(self.temp)
self.temp = []
self.tup = ibm_db.fetch_tuple(self.query_stmt)
except:
# self.error_list = 'WARNING: An unknown error occurred.'
self.temp = []
self.tup = []
def make_connection(self):
try:
# Connection
cnxn = ibm_db.connect(self.conn_string, '', '')
return cnxn
except:
self.error_list = 'WARNING: ' + ibm_db.conn_errormsg()
# messagebox.showinfo('Try','CON' + self.error_list)
def run_sql(self):
try:
# Connection
query_stmt = ibm_db.exec_immediate(self.cnxn, self.sql)
return query_stmt
except:
self.error_list = 'WARNING: ' + ibm_db.stmt_errormsg()
# messagebox.showinfo('Try','SQL' + self.error_list)
def get_data(self):
return self.q_list
def get_types(self):
return self.data_type
def get_headers(self):
return self.headers
def get_errorlist(self):
return self.error_list
