Source code for apache_beam.transforms.display

# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# 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('', url=''), '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)
[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_