Source code for pyflink.common.types

################################################################################
#  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.
################################################################################
from enum import Enum
from typing import List

from pyflink.java_gateway import get_gateway

__all__ = ['Row', 'RowKind']


[docs]class RowKind(Enum): INSERT = 0 UPDATE_BEFORE = 1 UPDATE_AFTER = 2 DELETE = 3 def __str__(self): if self.value == RowKind.INSERT.value: return '+I' elif self.value == RowKind.UPDATE_BEFORE.value: return '-U' elif self.value == RowKind.UPDATE_AFTER.value: return '+U' else: return '-D' def to_j_row_kind(self): JRowKind = get_gateway().jvm.org.apache.flink.types.RowKind return getattr(JRowKind, self.name)
def _create_row(fields, values, row_kind: RowKind = None): row = Row(*values) if fields is not None: row._fields = fields if row_kind is not None: row.set_row_kind(row_kind) return row
[docs]class Row(object): """ A row in Table. The fields in it can be accessed: * like attributes (``row.key``) * like dictionary values (``row[key]``) ``key in row`` will search through row keys. Row can be used to create a row object by using named arguments, the fields will be sorted by names. It is not allowed to omit a named argument to represent the value is None or missing. This should be explicitly set to None in this case. :: >>> row = Row(name="Alice", age=11) >>> row Row(age=11, name='Alice') >>> row['name'], row['age'] ('Alice', 11) >>> row.name, row.age ('Alice', 11) >>> 'name' in row True >>> 'wrong_key' in row False Row can also be used to create another Row like class, then it could be used to create Row objects, such as :: >>> Person = Row("name", "age") >>> Person <Row(name, age)> >>> 'name' in Person True >>> 'wrong_key' in Person False >>> Person("Alice", 11) Row(name='Alice', age=11) """ def __init__(self, *args, **kwargs): if args and kwargs: raise ValueError("Can not use both args " "and kwargs to create Row") if kwargs: self._fields = list(kwargs.keys()) self._values = [kwargs[n] for n in self._fields] self._from_dict = True else: self._values = list(args) self._row_kind = RowKind.INSERT def as_dict(self, recursive=False): """ Returns as a dict. Example: :: >>> Row(name="Alice", age=11).as_dict() == {'name': 'Alice', 'age': 11} True >>> row = Row(key=1, value=Row(name='a', age=2)) >>> row.as_dict() == {'key': 1, 'value': Row(age=2, name='a')} True >>> row.as_dict(True) == {'key': 1, 'value': {'name': 'a', 'age': 2}} True :param recursive: turns the nested Row as dict (default: False). """ if not hasattr(self, "_fields"): raise TypeError("Cannot convert a Row class into dict") if recursive: def conv(obj): if isinstance(obj, Row): return obj.as_dict(True) elif isinstance(obj, list): return [conv(o) for o in obj] elif isinstance(obj, dict): return dict((k, conv(v)) for k, v in obj.items()) else: return obj return dict(zip(self._fields, (conv(o) for o in self))) else: return dict(zip(self._fields, self)) def get_row_kind(self) -> RowKind: return self._row_kind def set_row_kind(self, row_kind: RowKind): self._row_kind = row_kind def set_field_names(self, field_names: List): self._fields = field_names def get_fields_by_names(self, names: List[str]): if not hasattr(self, '_fields') or names == self._fields: return self._values difference = list(set(names).difference(set(self._fields))) if difference: raise Exception("Field names {0} not exist in {1}.".format(difference, self._fields)) else: return [self._values[self._fields.index(name)] for name in names] def _is_retract_msg(self): return self._row_kind == RowKind.UPDATE_BEFORE or self._row_kind == RowKind.DELETE def _is_accumulate_msg(self): return self._row_kind == RowKind.UPDATE_AFTER or self._row_kind == RowKind.INSERT @staticmethod def of_kind(row_kind: RowKind, *args, **kwargs): row = Row(*args, **kwargs) row.set_row_kind(row_kind) return row def __contains__(self, item): return item in self._values # let object acts like class def __call__(self, *args): """ Creates new Row object """ if len(args) > len(self): raise ValueError("Can not create Row with fields %s, expected %d values " "but got %s" % (self, len(self), args)) return _create_row(self._values, args, self._row_kind) def __getitem__(self, item): if isinstance(item, (int, slice)): return self._values[item] try: # it will be slow when it has many fields, # but this will not be used in normal cases idx = self._fields.index(item) return self._values[idx] except IndexError: raise KeyError(item) except ValueError: raise ValueError(item) def __setitem__(self, key, value): if isinstance(key, (int, slice)): self._values[key] = value return try: # it will be slow when it has many fields, # but this will not be used in normal cases idx = self._fields.index(key) self._values[idx] = value except (IndexError, AttributeError): raise KeyError(key) except ValueError: raise ValueError(value) def __getattr__(self, item): if item.startswith("_"): raise AttributeError(item) try: # it will be slow when it has many fields, # but this will not be used in normal cases idx = self._fields.index(item) return self[idx] except IndexError: raise AttributeError(item) except ValueError: raise AttributeError(item) def __setattr__(self, key, value): if key != '_fields' and key != "_from_dict" and key != "_row_kind" and key != "_values": raise AttributeError(key) self.__dict__[key] = value def __reduce__(self): """ Returns a tuple so Python knows how to pickle Row. """ if hasattr(self, "_fields"): return _create_row, (self._fields, tuple(self), self._row_kind) else: return _create_row, (None, tuple(self), self._row_kind) def __repr__(self): """ Printable representation of Row used in Python REPL. """ if hasattr(self, "_fields"): return "Row(%s)" % ", ".join("%s=%r" % (k, v) for k, v in zip(self._fields, tuple(self))) else: return "<Row(%s)>" % ", ".join("%r" % field for field in self) def __eq__(self, other): if not isinstance(other, Row): return False if hasattr(self, "_fields"): if not hasattr(other, "_fields"): return False if sorted(self._fields) != sorted(other._fields): return False sorted_fields = sorted(self._fields) return (self.__class__ == other.__class__ and self._row_kind == other._row_kind and [self._values[self._fields.index(name)] for name in sorted_fields] == [other._values[other._fields.index(name)] for name in sorted_fields]) else: if hasattr(other, "_fields"): return False return (self.__class__ == other.__class__ and self._row_kind == other._row_kind and self._values == other._values) def __hash__(self): return tuple(self).__hash__() def __iter__(self): return iter(self._values) def __len__(self): return len(self._values)