Source code for deephaven.filters

#
# Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
#

""" This module implement various filters that can be used in deephaven table's filter operations."""
from __future__ import annotations

from enum import Enum
from typing import List, Union

import jpy
import functools

from deephaven import DHError
from deephaven._wrapper import JObjectWrapper
from deephaven.jcompat import to_sequence

_JFilter = jpy.get_type("io.deephaven.api.filter.Filter")
_JColumnName = jpy.get_type("io.deephaven.api.ColumnName")
_JFilterPattern = jpy.get_type("io.deephaven.api.filter.FilterPattern")
_JPatternMode = jpy.get_type("io.deephaven.api.filter.FilterPattern$Mode")
_JPattern = jpy.get_type("java.util.regex.Pattern")


[docs] class Filter(JObjectWrapper): """A Filter object represents a filter that can be used in Table's filtering(where) operations.""" j_object_type = _JFilter @property def j_object(self) -> jpy.JType: return self.j_filter def __init__(self, j_filter): self.j_filter = j_filter
[docs] def not_(self): """Creates a new filter that evaluates to the opposite of what this filter evaluates to. Returns: a new not Filter """ return Filter(j_filter=getattr(_JFilter, "not")(self.j_filter))
[docs] @classmethod def from_(cls, conditions: Union[str, List[str]]) -> Union[Filter, List[Filter]]: """Creates filter(s) from the given condition(s). Args: conditions (Union[str, List[str]]): filter condition(s) Returns: filter(s) Raises: DHError """ conditions = to_sequence(conditions) try: filters = [ cls(j_filter=j_filter) for j_filter in getattr(_JFilter, "from")(conditions).toArray() ] return filters if len(filters) != 1 else filters[0] except Exception as e: raise DHError(e, "failed to create filters.") from e
[docs] def or_(filters: Union[str, Filter, Sequence[str], Sequence[Filter]]) -> Filter: """Creates a new filter that evaluates to true when any of the given filters evaluates to true. Args: filters (Union[str, Filter, Sequence[str], Sequence[Filter]]): the component filter(s) Returns: a new or Filter """ seq = [ Filter.from_(f).j_filter if isinstance(f, str) else f for f in to_sequence(filters) ] return Filter(j_filter=getattr(_JFilter, "or")(*seq))
[docs] def and_(filters: Union[str, Filter, Sequence[str], Sequence[Filter]]) -> Filter: """Creates a new filter that evaluates to true when all of the given filters evaluates to true. Args: filters (Union[str, Filter, Sequence[str], Sequence[Filter]]): the component filters Returns: a new and Filter """ seq = [ Filter.from_(f).j_filter if isinstance(f, str) else f for f in to_sequence(filters) ] return Filter(j_filter=getattr(_JFilter, "and")(*seq))
[docs] def not_(filter_: Filter) -> Filter: """Creates a new filter that evaluates to the opposite of what filter_ evaluates to. Args: filter_ (Filter): the filter to negate with Returns: a new not Filter """ return filter_.not_()
[docs] def is_null(col: str) -> Filter: """Creates a new filter that evaluates to true when the col is null, and evaluates to false when col is not null. Args: col (str): the column name Returns: a new is-null Filter """ return Filter(j_filter=_JFilter.isNull(_JColumnName.of(col)))
[docs] def is_not_null(col: str) -> Filter: """Creates a new filter that evaluates to true when the col is not null, and evaluates to false when col is null. Args: col (str): the column name Returns: a new is-not-null Filter """ return Filter(j_filter=_JFilter.isNotNull(_JColumnName.of(col)))
[docs] class PatternMode(Enum): """The regex mode to use""" MATCHES = _JPatternMode.MATCHES """Matches the entire input against the pattern""" FIND = _JPatternMode.FIND """Matches any subsequence of the input against the pattern"""
[docs] def pattern( mode: PatternMode, col: str, regex: str, invert_pattern: bool = False ) -> Filter: """Creates a regular-expression pattern filter. See https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/regex/Pattern.html for documentation on the regex pattern. This filter will never match ``null`` values. Args: mode (PatternMode): the mode col (str): the column name regex (str): the regex pattern invert_pattern (bool): if the pattern matching logic should be inverted Returns: a new pattern filter Raises: DHError """ try: return Filter( j_filter=_JFilterPattern.of( _JColumnName.of(col), _JPattern.compile(regex), mode.value, invert_pattern, ) ) except Exception as e: raise DHError(e, "failed to create a pattern filter.") from e