# $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: Sort.py 10330 2009-12-24 22:13:38Z grisha $
"""
Sort Module and Resource Definition
"""
from snaplogic.common.snap_exceptions import SnapObjTypeError
import tempfile
import os
from decimal import Decimal
from datetime import datetime
import time
from sqlite3 import dbapi2
import snaplogic.components as components
from snaplogic.components import computils
from snaplogic.common import snap_log,sqlite_iter
from snaplogic.common import version_info
from snaplogic.cc import component_api
from snaplogic.cc.component_api import ComponentAPI
from snaplogic.cc import prop
from snaplogic.snapi_base import keys
from snaplogic.common.snap_exceptions import SnapComponentError
from snaplogic.common.data_types import SnapNumber,SnapString,SnapDateTime
DATE_FORMAT_STRING = "%Y-%m-%d %H:%M:%S"
FORMATTED_DATE_LENGTH = len(time.strftime(DATE_FORMAT_STRING, datetime.today().timetuple()))
SORT_FIELD = "Sort field"
SORT_ORDER = "Sort order"
SORT_SPEC = "Sort spec"
SORT_SPECS = "Sort specs"
class Sort(ComponentAPI):
"""
Sort.
"""
api_version = '1.0'
component_version = '1.3'
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,
}
component_description = "Sort"
component_label = "Sort"
component_doc_uri = "https://www.snaplogic.org/trac/wiki/Documentation/%s/ComponentRef/Sort" % \
version_info.doc_uri_version
def _adapt_Decimal(self, dec):
return str(dec)
def create_resource_template(self):
"""
Create Sort resource template.
"""
sort_field = prop.SimpleProp(SORT_FIELD, "string", "Input field to sort on",
{'lov': [ keys.CONSTRAINT_LOV_INPUT_FIELD] }, required=True)
sort_order = prop.SimpleProp(SORT_ORDER, "string", "Sort order (ASC or DESC)",
{ "lov" : [ "asc", "desc" ] }, required=True)
sort_spec = prop.DictProp(SORT_SPEC, sort_field, "describe me", 2, required=True)
sort_spec[SORT_FIELD] = sort_field
sort_spec[SORT_ORDER] = sort_order
sort_specs = prop.ListProp("Sort specifications", sort_spec, min_size = 1, required=True)
self.set_property_def(SORT_SPECS, sort_specs)
def validate(self, err_obj):
sort_specs = self.get_property_value(SORT_SPECS)
if component_api.has_param(sort_specs):
# Cannot do much more if this is still a parameter
return
# Validate that the output view is the same as the input view.
# Make sure that the output view matches the input view. (at least the types)
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_view_fields = input_view[keys.VIEW_FIELDS]
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 count matches?
if len(output_view[keys.VIEW_FIELDS]) != len(input_view_fields):
err_obj.get_output_view_err()[output_views[keys.SINGLE_VIEW]].set_message(
"Output view '%s' field count '%d' does not match corresponding input view '%s' field count '%d'." \
% (output_view_name, len(output_view[keys.VIEW_FIELDS]),
input_view_name, len(input_view_fields)))
else:
# Field types match?
for i, output_field in enumerate(output_view[keys.VIEW_FIELDS]):
output_field_name = output_field[keys.FIELD_NAME]
output_field_type = output_field[keys.FIELD_TYPE]
input_field_name = input_view_fields[i][keys.FIELD_NAME]
input_field_type = input_view_fields[i][keys.FIELD_TYPE]
if output_field_type != input_field_type:
err_obj.get_output_view_err()[output_views[keys.SINGLE_VIEW]][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, output_field_name, output_field_type,
input_view_name, input_field_name, input_field_type))
def execute(self, input_views, output_views):
try:
output_view = output_views.values()[keys.SINGLE_VIEW]
except IndexError:
raise SnapComponentError("No output view connected.")
try:
input_view = input_views.values()[keys.SINGLE_VIEW]
except IndexError:
raise SnapComponentError("No input view connected.")
fields_to_db_fields = {}
db_fields_to_fields = {}
field_types = {}
db_fields = []
fields = []
count = 0
input_view_fields = input_view.fields
for input_field in input_view_fields:
field = input_field[keys.FIELD_NAME]
field_type = input_field[keys.FIELD_TYPE]
field_types[field] = field_type
fields.append(field)
db_field = "field%s" % count
count += 1
db_fields_to_fields[db_field] = field
fields_to_db_fields[field] = db_field
sort_specs = self.get_property_value(SORT_SPECS)
sort_fields = []
sort_orders = {}
for spec in sort_specs:
sort_field = spec[SORT_FIELD]
sort_fields.append(sort_field)
sort_order = spec[SORT_ORDER]
sort_orders[sort_field] = sort_order
db_fields = [fields_to_db_fields[field] for field in fields]
db_file = None
db_file_name = None
cursor = None
con = None
try:
# Create a temp file to hold the SQLite database.
# Note that mkstemp opens the file as well, which we don't need,
# so close the temp file after it's been created.
(db_file, db_file_name) = tempfile.mkstemp(".db","snapsort")
os.close(db_file)
con = sqlite.connect(db_file_name)
cursor = con.cursor()
sqlite.register_adapter(Decimal, self._adapt_Decimal)
first = True
stmt = 'CREATE TABLE sort ('
for db_field in db_fields:
if not first:
stmt += ", "
stmt += db_field + " "
field = db_fields_to_fields[db_field]
field_type = field_types[field]
if field_type == SnapString:
stmt += "TEXT"
elif field_type == SnapNumber:
stmt += "DECIMAL"
elif field_type == SnapDateTime:
stmt += 'DATETIME'
else:
raise SnapObjTypeError('Unknown type %s', field_type)
if first:
first = False
stmt += ")"
print "Executing %s" % stmt
cursor.execute(stmt)
print "Done..."
insert_stmt = "INSERT INTO sort (" + \
",".join(db_fields) + \
") VALUES (" + \
",".join(['?' for i in db_fields]) + \
")"
self._process_records(input_view, cursor, insert_stmt)
stmt = "SELECT " + ",".join(db_fields) + " FROM sort ORDER BY "
first = True
for sort_field in sort_fields:
if first:
first = False
else:
stmt += ","
sort_db_field = fields_to_db_fields[sort_field]
stmt += sort_db_field
stmt += " " + sort_orders[sort_field]
cursor.execute(stmt)
output_view_fields = output_view.fields
output_field_types = [output_field[keys.FIELD_TYPE] for output_field in output_view_fields]
for row in sqlite_iter(cursor):
out_rec = output_view.create_record()
i = 0
for field in out_rec.field_names:
if row[i] is None:
out_rec[field] = row[i]
elif output_field_types[i] == SnapNumber:
out_rec[field] = Decimal(str(row[i]))
elif output_field_types[i] == SnapDateTime:
no_micros = row[i][:FORMATTED_DATE_LENGTH]
micros = row[i][FORMATTED_DATE_LENGTH+1:]
formatted = time.strptime(no_micros, DATE_FORMAT_STRING)
time_tuple = formatted[0:6]
dt = datetime(*time_tuple)
if micros:
dt = dt.replace(microsecond=int(micros))
out_rec[field] = dt
else:
out_rec[field] = row[i]
i += 1
output_view.write_record(out_rec)
output_view.completed()
except Exception, e:
# TODO
raise
finally:
if cursor:
try:
cursor.close()
except:
# Can't do much here
pass
if con:
try:
con.close()
except:
pass
if db_file_name:
try:
os.remove(db_file_name)
except:
pass
def _process_records(self, input_view, cursor, insert_stmt):
while True:
record = input_view.read_record()
if record is None:
break
vals = [record[field_name] for field_name in record.field_names]
cursor.execute(insert_stmt, vals)
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(SORT_SPECS)
sort_field = prop.SimpleProp(SORT_FIELD, "string", "Input field to sort on",
{'lov': [ keys.CONSTRAINT_LOV_INPUT_FIELD] }, required=True)
sort_order = prop.SimpleProp(SORT_ORDER, "string", "Sort order (ASC or DESC)",
{ "lov" : [ "asc", "desc" ] }, required=True)
sort_spec = prop.DictProp(SORT_SPEC, sort_field, "describe me", 2, required=True)
sort_spec[SORT_FIELD] = sort_field
sort_spec[SORT_ORDER] = sort_order
sort_specs = prop.ListProp("Sort specifications", sort_spec, required=True)
self.set_property_def(SORT_SPECS, sort_specs)
# Restore the value
self.set_property_value(SORT_SPECS, property_value)
def upgrade_1_1_to_1_2(self):
""" Add size constraint on sort specs """
# Save the current value of sort specs
saved_sort_specs = self.get_property_value(SORT_SPECS)
sort_field = prop.SimpleProp(SORT_FIELD, "string", "Input field to sort on",
{'lov': [ keys.CONSTRAINT_LOV_INPUT_FIELD] }, required=True)
sort_order = prop.SimpleProp(SORT_ORDER, "string", "Sort order (ASC or DESC)",
{ "lov" : [ "asc", "desc" ] }, required=True)
sort_spec = prop.DictProp(SORT_SPEC, sort_field, "describe me", 2, required=True)
sort_spec[SORT_FIELD] = sort_field
sort_spec[SORT_ORDER] = sort_order
sort_specs = prop.ListProp("Sort specifications", sort_spec, min_size = 1, required=True)
self.set_property_def(SORT_SPECS, sort_specs)
# Restore the sort specs
self.set_property_value(SORT_SPECS, saved_sort_specs)
def upgrade_1_2_to_1_3(self):
"""
No-op upgrade only to change component doc URI during the upgrade
which will be by cc_info before calling this method.
"""
pass
|