taps.data.filter

taps/data/filter.py

Filter

Bases: Protocol

Filter protocol.

__call__()

__call__(obj: Any) -> bool

Check if an abject passes through the filter.

Source code in taps/data/filter.py

def __call__(self, obj: Any) -> bool:
    """Check if an abject passes through the filter."""
    ...

NullFilter

Null filter that lets all objects pass through.

__call__()

__call__(obj: Any) -> bool

Check if an object passes through the filter.

Source code in taps/data/filter.py

def __call__(self, obj: Any) -> bool:
    """Check if an object passes through the filter."""
    return True

ObjectSizeFilter

ObjectSizeFilter(
    *, min_bytes: int = 0, max_bytes: float = math.inf
)

Object size filter.

Checks if the size of an object (computed using sys.getsizeof()) is greater than a minimum size and less than a maximum size.

Warning

sys.getsizeof() does not count the size of objects referred to by the main object.

Parameters:

• min_bytes (int, default: 0 ) –

  Minimum size threshold (inclusive) to pass through the filter.

• max_bytes (float, default: inf ) –

  Maximum size threshold (inclusive) to pass through the filter.

Source code in taps/data/filter.py

def __init__(
    self,
    *,
    min_bytes: int = 0,
    max_bytes: float = math.inf,
) -> None:
    self.min_bytes = min_bytes
    self.max_bytes = max_bytes

__call__()

__call__(obj: Any) -> bool

Check if an object passes through the filter.

Source code in taps/data/filter.py

def __call__(self, obj: Any) -> bool:
    """Check if an object passes through the filter."""
    size = sys.getsizeof(obj)
    return self.min_bytes <= size <= self.max_bytes

ObjectTypeFilter

ObjectTypeFilter(*types: type)

Object type filter.

Checks if an object is of a certain type.

Parameters:

• types (type, default: () ) –

  Types to check.

Source code in taps/data/filter.py

def __init__(self, *types: type) -> None:
    self.types = types

__call__()

__call__(obj: Any) -> bool

Check if an object passes through the filter.

Source code in taps/data/filter.py

def __call__(self, obj: Any) -> bool:
    """Check if an object passes through the filter."""
    return isinstance(obj, self.types)

PickleSizeFilter

PickleSizeFilter(
    *, min_bytes: int = 0, max_bytes: float = math.inf
)

Object size filter.

Checks if the size of an object (computed using size of the pickled object) is greater than a minimum size and less than a maximum size.

Warning

Pickling large objects can take significant time.

Parameters:

• min_bytes (int, default: 0 ) –

  Minimum size threshold (inclusive) to pass through the filter.

• max_bytes (float, default: inf ) –

  Maximum size threshold (inclusive) to pass through the filter.

Source code in taps/data/filter.py

def __init__(
    self,
    *,
    min_bytes: int = 0,
    max_bytes: float = math.inf,
) -> None:
    self.min_bytes = min_bytes
    self.max_bytes = max_bytes

__call__()

__call__(obj: Any) -> bool

Check if an object passes through the filter.

Source code in taps/data/filter.py

def __call__(self, obj: Any) -> bool:
    """Check if an object passes through the filter."""
    size = len(pickle.dumps(obj))
    return self.min_bytes <= size <= self.max_bytes