#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""
DisplayData, its classes, interfaces and methods.
The classes in this module allow users and transform developers to define
static display data to be displayed when a pipeline runs. PTransforms, DoFns
and other pipeline components are subclasses of the HasDisplayData mixin. To
add static display data to a component, you can override the display_data
method of the HasDisplayData class.
Available classes:
- HasDisplayData - Components that inherit from this class can have static
display data shown in the UI.
- DisplayDataItem - This class represents static display data elements.
- DisplayData - Internal class that is used to create display data and
communicate it to the API.
"""
from __future__ import absolute_import
import calendar
from datetime import datetime, timedelta
import inspect
import json
__all__ = ['HasDisplayData', 'DisplayDataItem', 'DisplayData']
[docs]class HasDisplayData(object):
""" Basic mixin for elements that contain display data.
It implements only the display_data method and a _namespace method.
"""
[docs] def display_data(self):
""" Returns the display data associated to a pipeline component.
It should be reimplemented in pipeline components that wish to have
static display data.
Returns:
A dictionary containing key:value pairs. The value might be an
integer, float or string value; a DisplayDataItem for values that
have more data (e.g. short value, label, url); or a HasDisplayData
instance that has more display data that should be picked up. For
example:
{ 'key1': 'string_value',
'key2': 1234,
'key3': 3.14159265,
'key4': DisplayDataItem('apache.org', url='http://apache.org'),
'key5': subComponent }
"""
return {}
def _namespace(self):
return '{}.{}'.format(self.__module__, self.__class__.__name__)
[docs]class DisplayData(object):
""" Static display data associated with a pipeline component.
"""
def __init__(self, namespace, display_data_dict):
self.namespace = namespace
self.items = []
self._populate_items(display_data_dict)
def _populate_items(self, display_data_dict):
""" Populates the list of display data items.
"""
for key, element in display_data_dict.items():
if isinstance(element, HasDisplayData):
subcomponent_display_data = DisplayData(element._namespace(),
element.display_data())
self.items += subcomponent_display_data.items
continue
if isinstance(element, DisplayDataItem):
if element.should_drop():
continue
element.key = key
element.namespace = self.namespace
self.items.append(element)
continue
# If it's not a HasDisplayData element,
# nor a dictionary, then it's a simple value
self.items.append(
DisplayDataItem(element,
namespace=self.namespace,
key=key))
@classmethod
[docs] def create_from_options(cls, pipeline_options):
""" Creates DisplayData from a PipelineOptions instance.
When creating DisplayData, this method will convert the value of any
item of a non-supported type to its string representation.
The normal DisplayData.create_from method rejects those items.
Returns:
A DisplayData instance with populated items.
Raises:
ValueError: If the has_display_data argument is not an instance of
HasDisplayData.
"""
from apache_beam.options.pipeline_options import PipelineOptions
if not isinstance(pipeline_options, PipelineOptions):
raise ValueError(
'Element of class {}.{} does not subclass PipelineOptions'
.format(pipeline_options.__module__,
pipeline_options.__class__.__name__))
items = {k: (v if DisplayDataItem._get_value_type(v) is not None
else str(v))
for k, v in pipeline_options.display_data().items()}
return cls(pipeline_options._namespace(), items)
@classmethod
[docs] def create_from(cls, has_display_data):
""" Creates DisplayData from a HasDisplayData instance.
Returns:
A DisplayData instance with populated items.
Raises:
ValueError: If the has_display_data argument is not an instance of
HasDisplayData.
"""
if not isinstance(has_display_data, HasDisplayData):
raise ValueError('Element of class {}.{} does not subclass HasDisplayData'
.format(has_display_data.__module__,
has_display_data.__class__.__name__))
return cls(has_display_data._namespace(), has_display_data.display_data())
[docs]class DisplayDataItem(object):
""" A DisplayDataItem represents a unit of static display data.
Each item is identified by a key and the namespace of the component the
display item belongs to.
"""
typeDict = {str:'STRING',
unicode:'STRING',
int:'INTEGER',
float:'FLOAT',
bool: 'BOOLEAN',
timedelta:'DURATION',
datetime:'TIMESTAMP'}
def __init__(self, value, url=None, label=None,
namespace=None, key=None, shortValue=None):
self.namespace = namespace
self.key = key
self.type = self._get_value_type(value)
self.shortValue = (shortValue if shortValue is not None else
self._get_short_value(value, self.type))
self.value = value
self.url = url
self.label = label
self._drop_if_none = False
self._drop_if_default = False
[docs] def drop_if_none(self):
""" The item should be dropped if its value is None.
Returns:
Returns self.
"""
self._drop_if_none = True
return self
[docs] def drop_if_default(self, default):
""" The item should be dropped if its value is equal to its default.
Returns:
Returns self.
"""
self._default = default
self._drop_if_default = True
return self
[docs] def should_drop(self):
""" Return True if the item should be dropped, or False if it should not
be dropped. This depends on the drop_if_none, and drop_if_default calls.
Returns:
True or False; depending on whether the item should be dropped or kept.
"""
if self._drop_if_none and self.value is None:
return True
if self._drop_if_default and self.value == self._default:
return True
return False
[docs] def is_valid(self):
""" Checks that all the necessary fields of the DisplayDataItem are
filled in. It checks that neither key, namespace, value or type are None.
Raises:
ValueError: If the item does not have a key, namespace, value or type.
"""
if self.key is None:
raise ValueError('Invalid DisplayDataItem. Key must not be None')
if self.namespace is None:
raise ValueError('Invalid DisplayDataItem. Namespace must not be None')
if self.value is None:
raise ValueError('Invalid DisplayDataItem. Value must not be None')
if self.type is None:
raise ValueError(
'Invalid DisplayDataItem. Value {} is of an unsupported type.'
.format(self.value))
def _get_dict(self):
res = {'key': self.key,
'namespace': self.namespace,
'type': self.type if self.type != 'CLASS' else 'STRING'}
# TODO: Python Class types should not be special-cased once
# the Fn API is in.
if self.url is not None:
res['url'] = self.url
if self.shortValue is not None:
res['shortValue'] = self.shortValue
if self.label is not None:
res['label'] = self.label
res['value'] = self._format_value(self.value, self.type)
return res
[docs] def get_dict(self):
""" Returns the internal-API dictionary representing the DisplayDataItem.
Returns:
A dictionary. The internal-API dictionary representing the
DisplayDataItem
Raises:
ValueError: if the item is not valid.
"""
self.is_valid()
return self._get_dict()
def __repr__(self):
return 'DisplayDataItem({})'.format(json.dumps(self._get_dict()))
def __eq__(self, other):
if isinstance(other, self.__class__):
return self._get_dict() == other._get_dict()
return False
def __ne__(self, other):
return not self == other
def __hash__(self):
return hash(tuple(sorted(self._get_dict().items())))
@classmethod
def _format_value(cls, value, type_):
""" Returns the API representation of a value given its type.
Args:
value: The value of the item that needs to be shortened.
type_(string): The type of the value.
Returns:
A formatted value in the form of a float, int, or string.
"""
res = value
if type_ == 'CLASS':
res = '{}.{}'.format(value.__module__, value.__name__)
elif type_ == 'DURATION':
res = value.total_seconds()*1000
elif type_ == 'TIMESTAMP':
res = calendar.timegm(value.timetuple())*1000 + value.microsecond//1000
return res
@classmethod
def _get_short_value(cls, value, type_):
""" Calculates the short value for an item.
Args:
value: The value of the item that needs to be shortened.
type_(string): The type of the value.
Returns:
The unqualified name of a class if type_ is 'CLASS'. None otherwise.
"""
if type_ == 'CLASS':
return value.__name__
return None
@classmethod
def _get_value_type(cls, value):
""" Infers the type of a given value.
Args:
value: The value whose type needs to be inferred. For 'DURATION' and
'TIMESTAMP', the corresponding Python type is datetime.timedelta and
datetime.datetime respectively. For Python classes, the API type is
just 'STRING' at the moment.
Returns:
One of 'STRING', 'INTEGER', 'FLOAT', 'CLASS', 'DURATION', or
'TIMESTAMP', depending on the type of the value.
"""
#TODO: Fix Args: documentation once the Python classes handling has changed
type_ = cls.typeDict.get(type(value))
if type_ is None:
type_ = 'CLASS' if inspect.isclass(value) else None
if type_ is None and value is None:
type_ = 'STRING'
return type_