A Bloom filter is a space-efficient probabilistic data structure, conceived by Burton Howard Bloom in 1970, that is used to test whether an element is a member of a set.

bloom_filter.py

"""Bloom Filter

A Bloom filter is a space-efficient probabilistic data structure, 
conceived by Burton Howard Bloom in 1970, that is used to test whether 
an element is a member of a set. False positive matches are possible, 
but false negatives are not – in other words, a query returns either 
"possibly in set" or "definitely not in set". Elements can be added to 
the set, but not removed (though this can be addressed with the counting 
Bloom filter variant); the more items added, the larger the probability 
of false positives.

Bloom proposed the technique for applications where the amount of source 
data would require an impractically large amount of memory if "conventional" 
error-free hashing techniques were applied. He gave the example of a hyphenation 
algorithm for a dictionary of 500,000 words, out of which 90% follow simple 
hyphenation rules, but the remaining 10% require expensive disk accesses to 
retrieve specific hyphenation patterns. With sufficient core memory, an error-free 
hash could be used to eliminate all unnecessary disk accesses; on the other hand,
with limited core memory, Bloom's technique uses a smaller hash area but still 
eliminates most unnecessary accesses. For example, a hash area only 15% of the 
size needed by an ideal error-free hash still eliminates 85% of the disk accesses.

More generally, fewer than 10 bits per element are required for a 1% false positive 
probability, independent of the size or number of elements in the set.

source: https://en.wikipedia.org/wiki/Bloom_filter
author: https://github.com/ddjerqq
"""

import math
from typing import (
    TypeVar,
    Generic,
)

try:
    import mmh3
except ImportError as e:
    raise ImportError("Please install mmh3 module: pip install mmh3") from e

try:
    import bitarray
except ImportError as e:
    raise ImportError("Please install bitarray module: pip install bitarray") from e


T = TypeVar("T")


class BloomFilter(Generic[T]):
    """Bloom filter

    A probabilistic data structure to test whether an element is a member of a set.
    The more elements added, the larger the probability of false positives.
    False positives mean that the element is not in the set, but the filter says it is.
    But if the filter says an element is *NOT* in the set, then it is definitely *NOT* in the set.
    """

    def __init__(self, item_count: int, max_allowed_false_positive_probability: float) -> None:
        """Create a new Bloom filter

        Args:
            item_count: number of items to be stored in the filter
            max_allowed_false_positive_probability: maximum allowed false positive probability
        """
        self._fp_prob = max_allowed_false_positive_probability
        """float: Maximum allowed false positive probability"""

        self._size = self._get_size(item_count, max_allowed_false_positive_probability)
        """int: Size of bit array to use"""

        self._hash_count = self._get_hash_count(self._size, item_count)
        """int: number of hash functions to use"""

        self._bit_array = bitarray.bitarray(self._size)
        """bitarray: Bit array of given size"""

        self._bit_array.setall(0)


    @staticmethod
    def _get_size(n: int, p: float) -> int:
        """Returns the size of bit array(m) to used using

        m = -(n * lg(p)) / (lg(2)^2)

        Args:
            n: item_count is the number of items expected to be stored in filter
            p: max_allowed_false_positive_probability is the maximum tolerable false positive probability
        """
        m = -(n * math.log(p)) / (math.log(2) ** 2)
        return int(m)

    
    @staticmethod
    def _get_hash_count(m: int, n: int) -> int:
        """Returns the hash function(k) to be used using

        k = (m/n) * lg(2)

        Args:
            m: size of bit array
            n: item_count is the number of items expected to be stored in filter
        """
        k = (m / n) * math.log(2)
        return int(k)


    def add(self, item: T) -> None:
        """Add an item to the filter

        Args:
            item: item to add
        """
        digests = []
        for i in range(self._hash_count):
            # create digest for given item.
            # i works as a seed to mmh3.hash() function
            # With different seed, digest created is different
            digest = mmh3.hash(item, i) % self._size
            digests.append(digest)

            # set the bit True in bit_array
            self._bit_array[digest] = True

    
    def contains(self, item: T) -> bool:
        """Check for existence of an item in filter

        Args:
            item: item to check

        Returns:
            bool: True if item exists in filter, False otherwise
        """
        for i in range(self._hash_count):
            digest = mmh3.hash(item, i) % self._size
            if self._bit_array[digest] == False:
                # if any of bit is False then, its not present in filter
                # else there is probability that it exist
                return False
        return True


    def __contains__(self, item: T) -> bool:
        """Check for existence of an item in filter

        Args:
            item: item to check

        Returns:
            bool: True if item exists in filter, False otherwise
        """
        return self.contains(item)

    
    def __len__(self) -> int:
        """Returns the size of bit array"""
        return self._size


def test_bloom_filter():
    """Test Bloom filter"""
    # create a Bloom filter with expected item count of 1000 and
    # max allowed false positive probability of 0.1
    bloom_filter = BloomFilter(1000, 0.1)

    # add some items to filter
    bloom_filter.add("dante")
    bloom_filter.add("buddha")
    bloom_filter.add("turing")
    bloom_filter.add("einstein")
    bloom_filter.add("bohr")
    bloom_filter.add("darwin")
    bloom_filter.add("pasteur")

    # check for existence of an item
    assert "dante" in bloom_filter
    assert "buddha" in bloom_filter
    assert "turing" in bloom_filter
    assert "einstein" in bloom_filter
    assert "bohr" in bloom_filter
    assert "darwin" in bloom_filter
    assert "pasteur" in bloom_filter

    # check for existence of an item which is not added
    assert "tesla" not in bloom_filter
    assert "edison" not in bloom_filter
    assert "bell" not in bloom_filter
    assert "faraday" not in bloom_filter
    assert "maxwell" not in bloom_filter
    assert "galileo" not in bloom_filter
    assert "newton" not in bloom_filter
    assert "euclid" not in bloom_filter


if __name__ == "__main__":
    test_bloom_filter()