#
# 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.
#
"""
:class:`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.
:class:`~apache_beam.transforms.ptransform.PTransform` s,
:class:`~apache_beam.transforms.core.DoFn` s
and other pipeline components are subclasses of the :class:`HasDisplayData`
mixin. To add static display data to a component, you can override the
:meth:`HasDisplayData.display_data()` method.
Available classes:
* :class:`HasDisplayData` - Components that inherit from this class can have
static display data shown in the UI.
* :class:`DisplayDataItem` - This class represents static display data
elements.
* :class:`DisplayData` - Internal class that is used to create display data
and communicate it to the API.
"""
# pytype: skip-file
import calendar
import inspect
import json
from datetime import datetime
from datetime import timedelta
from typing import TYPE_CHECKING
from typing import List
from typing import Optional
from typing import Union
from apache_beam.portability import common_urns
from apache_beam.portability.api import beam_runner_api_pb2
if TYPE_CHECKING:
from apache_beam.options.pipeline_options import PipelineOptions
__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
_get_display_data_namespace method.
"""
[docs] def display_data(self):
# type: () -> dict
""" Returns the display data associated to a pipeline component.
It should be reimplemented in pipeline components that wish to have
static display data.
Returns:
Dict[str, Any]: A dictionary containing ``key:value`` pairs.
The value might be an integer, float or string value; a
:class:`DisplayDataItem` for values that have more data
(e.g. short value, label, url); or a :class:`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 _get_display_data_namespace(self):
# type: () -> str
return '{}.{}'.format(self.__module__, self.__class__.__name__)
[docs]class DisplayData(object):
""" Static display data associated with a pipeline component.
"""
def __init__(
self,
namespace, # type: str
display_data_dict # type: dict
):
# type: (...) -> None
self.namespace = namespace
self.items = [
] # type: List[Union[DisplayDataItem, beam_runner_api_pb2.DisplayData]]
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._get_display_data_namespace(), element.display_data())
self.items += subcomponent_display_data.items
elif isinstance(element, DisplayDataItem):
if element.should_drop():
continue
element.key = key
element.namespace = self.namespace
self.items.append(element)
elif isinstance(element, beam_runner_api_pb2.DisplayData):
self.items.append(element)
else:
# 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))
[docs] def to_proto(self):
# type: (...) -> List[beam_runner_api_pb2.DisplayData]
"""Returns a List of Beam proto representation of Display data."""
def create_payload(dd) -> Optional[beam_runner_api_pb2.LabelledPayload]:
try:
display_data_dict = dd.get_dict()
except ValueError:
# Skip if the display data is invalid.
return None
# We use 'label' or 'key' properties to populate the 'label' attribute of
# 'LabelledPayload'. 'label' is a better choice since it's expected to be
# more human readable but some transforms, sources, etc. may not set a
# 'label' property when configuring DisplayData.
label = (
display_data_dict['label']
if 'label' in display_data_dict else display_data_dict['key'])
value = display_data_dict['value']
if isinstance(value, str):
return beam_runner_api_pb2.LabelledPayload(
label=label,
string_value=value,
key=display_data_dict['key'],
namespace=display_data_dict.get('namespace', ''))
elif isinstance(value, bool):
return beam_runner_api_pb2.LabelledPayload(
label=label,
bool_value=value,
key=display_data_dict['key'],
namespace=display_data_dict.get('namespace', ''))
elif isinstance(value, int):
return beam_runner_api_pb2.LabelledPayload(
label=label,
int_value=value,
key=display_data_dict['key'],
namespace=display_data_dict.get('namespace', ''))
elif isinstance(value, (float, complex)):
return beam_runner_api_pb2.LabelledPayload(
label=label,
double_value=value,
key=display_data_dict['key'],
namespace=display_data_dict.get('namespace', ''))
else:
raise ValueError(
'Unsupported type %s for value of display data %s' %
(type(value), label))
dd_protos = []
for dd in self.items:
if isinstance(dd, beam_runner_api_pb2.DisplayData):
dd_protos.append(dd)
else:
dd_payload = create_payload(dd)
if dd_payload:
dd_protos.append(
beam_runner_api_pb2.DisplayData(
urn=common_urns.StandardDisplayData.DisplayData.LABELLED.urn,
payload=dd_payload.SerializeToString()))
return dd_protos
[docs] @classmethod
def create_from_options(cls, pipeline_options):
""" Creates :class:`~apache_beam.transforms.display.DisplayData` from a
:class:`~apache_beam.options.pipeline_options.PipelineOptions` instance.
When creating :class:`~apache_beam.transforms.display.DisplayData`, this
method will convert the value of any item of a non-supported type to its
string representation.
The normal :meth:`.create_from()` method rejects those items.
Returns:
~apache_beam.transforms.display.DisplayData:
A :class:`~apache_beam.transforms.display.DisplayData` instance with
populated items.
Raises:
ValueError: If the **has_display_data** argument is
not an instance of :class:`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._get_display_data_namespace(), items)
[docs] @classmethod
def create_from(cls, has_display_data):
""" Creates :class:`~apache_beam.transforms.display.DisplayData` from a
:class:`HasDisplayData` instance.
Returns:
~apache_beam.transforms.display.DisplayData:
A :class:`~apache_beam.transforms.display.DisplayData` instance with
populated items.
Raises:
ValueError: If the **has_display_data** argument is
not an instance of :class:`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._get_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',
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):
# type: () -> DisplayDataItem
""" 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):
# type: (...) -> DisplayDataItem
""" 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):
# type: () -> bool
""" 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):
# type: () -> None
""" Checks that all the necessary fields of the :class:`DisplayDataItem`
are filled in. It checks that neither key, namespace, value or type are
:data:`None`.
Raises:
ValueError: If the item does not have a key, namespace,
value or type.
"""
if self.key is None:
raise ValueError(
'Invalid DisplayDataItem %s. Key must not be None.' % self)
if self.namespace is None:
raise ValueError(
'Invalid DisplayDataItem %s. Namespace must not be None' % self)
if self.value is None:
raise ValueError(
'Invalid DisplayDataItem %s. Value must not be None' % self)
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):
# type: () -> dict
""" Returns the internal-API dictionary representing the
:class:`DisplayDataItem`.
Returns:
Dict[str, Any]: A dictionary. The internal-API dictionary representing
the :class:`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 __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_