Bloom Filter With Double Hashing

Size a Bloom filter from an expected item count and false-positive rate, then add and test members using double hashing.

PythonData Integrity

Code

snippet
from __future__ import annotations

import hashlib
import math


class BloomFilter:
    def __init__(
        self,
        expected_items: int,
        false_positive_rate: float = 0.01,
    ) -> None:
        if (
            not isinstance(expected_items, int)
            or isinstance(expected_items, bool)
            or expected_items < 1
        ):
            raise ValueError("expected_items must be a positive integer")
        if (
            not isinstance(false_positive_rate, (int, float))
            or isinstance(false_positive_rate, bool)
            or not 0 < false_positive_rate < 1
        ):
            raise ValueError("false_positive_rate must be between 0 and 1")

        self.expected_items = expected_items
        self.false_positive_rate = float(false_positive_rate)
        self.size = max(
            1,
            math.ceil(
                -(expected_items * math.log(self.false_positive_rate))
                / (math.log(2) ** 2)
            ),
        )
        self.hash_count = max(
            1,
            round((self.size / expected_items) * math.log(2)),
        )
        self._bits = bytearray((self.size + 7) // 8)
        self.added = 0

    @staticmethod
    def _require_text(value: str) -> str:
        if not isinstance(value, str):
            raise TypeError("value must be a string")
        if not value:
            raise ValueError("value must not be empty")
        return value

    def _positions(self, value: str):
        digest = hashlib.sha256(value.encode("utf-8")).digest()
        first = int.from_bytes(digest[:8], "big")
        second = int.from_bytes(digest[8:16], "big") or 1
        for index in range(self.hash_count):
            yield (first + index * second) % self.size

    def add(self, value: str) -> None:
        for position in self._positions(self._require_text(value)):
            self._bits[position // 8] |= 1 << (position % 8)
        self.added += 1

    def __contains__(self, value: str) -> bool:
        text = self._require_text(value)
        return all(
            self._bits[position // 8] & (1 << (position % 8))
            for position in self._positions(text)
        )

    @property
    def fill_ratio(self) -> float:
        bits_set = sum(byte.bit_count() for byte in self._bits)
        return bits_set / self.size


def main() -> None:
    bloom = BloomFilter(expected_items=1000, false_positive_rate=0.01)
    for word in ["alpha", "beta", "gamma", "delta"]:
        bloom.add(word)

    print(all(word in bloom for word in ["alpha", "beta", "gamma", "delta"]))
    print("epsilon" in bloom)
    print(bloom.size, bloom.hash_count, round(bloom.fill_ratio, 4))

    try:
        BloomFilter(0)
    except ValueError as error:
        print(error)


if __name__ == "__main__":
    main()