# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 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: data_types.py 8009 2009-06-26 04:30:52Z grisha $
"""
Provides handling of Snap data types and records.
As of Beale, ASN.1 handling code is no longer here, it has been migrated to
L{snaplogic.rp.Asn1RP}.
"""
__docformat__ = "epytext en"
from datetime import date,datetime
from decimal import Decimal
from types import NoneType
from UserList import UserList
from weakref import WeakKeyDictionary
from snaplogic.common.snap_exceptions import SnapEOFError,SnapObjTypeError,SnapIOError,SnapException
import snaplogic.common.snap_exceptions as SnapExceptions
SnapString = 'string'
SnapNumber = 'number'
SnapDateTime = 'datetime'
"""Maps user-visible Snap types to internal Python types."""
snap_to_python_types = {
SnapString : unicode,
SnapNumber : Decimal,
SnapDateTime : datetime
}
"""List of allowed Snap types."""
snap_types = snap_to_python_types.keys()
class _FieldDict(dict):
"""
Represents field name-field value dictionary, only allowing
valid types to be set for each field. Performs automatic
coercion when necessary, so that str is stored as unicode,
float, long and int as Decimal, and date as datetime.
"""
def __init__(self, fieldTypes):
"""Initialize this dictionary, setting all fields to None."""
self.__fieldTypes = fieldTypes
# set default values
for field in self.__fieldTypes.keys():
self.__setitem__(field, None)
def __setitem__(self, name, value):
"""
Set a value for a field. The value is automatically coerced
into the appropriate type as follows: str to unicode,
float, int and long into Decimal, and date into datetime.
@param name: Field name
@type name: str
@param value: Field value
@type value: One of: unicode, str, float, int, long, Decimal,
date, datetime or NoneType
@raise SnapObjTypeError: if the value supplied is not one of the
allowed types, or is not allowed for the given field.
"""
t = type(value)
if t == str:
value = value.decode('utf-8')
if t == date:
value = datetime(value.year, value.month, value.day)
if t == float or t == int or t == long:
value = Decimal(str(value))
# Now we've done the needed conversions...
# Is this type even allowed?
t = type(value)
if not (t in [datetime, unicode, Decimal, NoneType]):
raise SnapObjTypeError("Cannot handle type: %s" % str(t))
# Is this type allowed for the field?
if t != NoneType and self.__fieldTypes[name] != t:
raise SnapObjTypeError("Cannot assign %s (%s) to %s field %s"
% (value, t, self.__fieldTypes[name], name))
dict.__setitem__(self, name, value)
class Record(dict):
"""
The class represents a data record.
This class represents the record objects that are returned when a record mode input view
is read. It is also the object that is passed to write_record() method of an output view
while writing out a record.
The record contains the name
"""
def __init__(self, view_name, field_names, python_types_map):
"""
Construct the record object.
@param view_name: The name of input or output view, whose field structure, this
record represents.
@type view_name: str
@param field_names: List of field names in that view.
@type field_names: list of str
@param python_types_map: Dictionary mapping the field names to their corresponding python type.
@type python_types_map: dict
"""
self._internal_pass_through = {}
self._external_pass_through = {}
self.field_names = field_names
self.python_types_map = python_types_map
self.view_name = view_name
@classmethod
def create_python_types_map(cls, field_names, field_types):
"""
Create mapping from field name to python types.
The return value of this method is used to provide the python types mape that is
typically used in instantiating Record objects.
@param field_names: List of field names.
@type field_names: list of str
@param field_types: List of SnapLogic field types.
@type field_types: list of str
@return: Dictionary mapping the field names to their corresponding python type.
@rtype: dict
"""
py_types = {}
for i in range(len(field_names)):
try:
py_types[field_names[i]] = snap_to_python_types[field_types[i]]
except KeyError:
raise SnapObjTypeError("Invalid type: %s" % field_types[i])
return py_types
def transfer_matching_fields(self, input_rec, fields):
"""
Transfer fields from one record to another by name.
This is useful for copying fields from input view to output view.
E.g. if input view has fields (a, b, c)
and output view has fields (b, c, d)
this function will copy fields b and c from input record to output record.
"""
for field in fields:
self[field] = input_rec[field]
def transfer_pass_through_fields(self, input_rec):
"""
Append pass-through fields from the input record to this record.
This method should be called to append pass-through fields from
an input view record to an output view record. The appending of
input view records must be called in the order specified by the
pass-through settings for that output view.
@param input_rec: Input record.
@type input_rec: L{Record}
"""
if input_rec.view_name in input_rec._internal_pass_through:
self._internal_pass_through[input_rec.view_name] = input_rec._internal_pass_through[input_rec.view_name]
self._external_pass_through[input_rec.view_name] = input_rec._external_pass_through[input_rec.view_name]
def dump_pass_through_fields(self):
"""
Return all the pass-through fields in the record.
This method should only be called in sink component that intends to
handle the pass through fields, only to dump it into some external
sink.
@return: List of pass through field values.
@rtype: list
"""
if self.view_name in self._internal_pass_through:
return self._internal_pass_through[self.view_name] + self._external_pass_through[self.view_name]
else:
return []
|