Add hash map

data_structures/hashing/hash_map.py

@@ -0,0 +1,162 @@
+"""
+Hash map with open addressing.
+
+https://en.wikipedia.org/wiki/Hash_table
+
+Another hash map implementation, with a good explanation.
+Modern Dictionaries by Raymond Hettinger
+https://www.youtube.com/watch?v=p33CVV29OG8
+"""
+from collections.abc import Iterator, MutableMapping
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+KEY = TypeVar("KEY")
+VAL = TypeVar("VAL")
+
+
+@dataclass(frozen=True, slots=True)
+class _Item(Generic[KEY, VAL]):
+    key: KEY
+    val: VAL
+
+
+class _DeletedItem(_Item):
+    def __init__(self) -> None:
+        super().__init__(None, None)
+
+    def __bool__(self) -> bool:
+        return False
+
+
+_deleted = _DeletedItem()
+
+
+class HashMap(MutableMapping[KEY, VAL]):
+    """
+    Hash map with open addressing.
+    """
+
+    def __init__(
+        self, initial_block_size: int = 8, capacity_factor: float = 0.75
+    ) -> None:
+        self._initial_block_size = initial_block_size
+        self._buckets: list[_Item | None] = [None] * initial_block_size
+        assert 0.0 < capacity_factor < 1.0
+        self._capacity_factor = capacity_factor
+        self._len = 0
+
+    def _get_bucket_index(self, key: KEY) -> int:
+        return hash(key) % len(self._buckets)
+
+    def _get_next_ind(self, ind: int) -> int:
+        """
+        Get next index.
+
+        Implements linear open addressing.
+        """
+        return (ind + 1) % len(self._buckets)
+
+    def _try_set(self, ind: int, key: KEY, val: VAL) -> bool:
+        """
+        Try to add value to the bucket.
+
+        If bucket is empty or key is the same, does insert and return True.
+
+        If bucket has another key or deleted placeholder,
+        that means that we need to check next bucket.
+        """
+        stored = self._buckets[ind]
+        if not stored:
+            self._buckets[ind] = _Item(key, val)
+            self._len += 1
+            return True
+        elif stored.key == key:
+            self._buckets[ind] = _Item(key, val)
+            return True
+        else:
+            return False
+
+    def _is_full(self) -> bool:
+        """
+        Return true if we have reached safe capacity.
+
+        So we need to increase the number of buckets to avoid collisions.
+        """
+        limit = len(self._buckets) * self._capacity_factor
+        return len(self) >= int(limit)
+
+    def _is_sparse(self) -> bool:
+        """Return true if we need twice fewer buckets when we have now."""
+        if len(self._buckets) <= self._initial_block_size:
+            return False
+        limit = len(self._buckets) * self._capacity_factor / 2
+        return len(self) < limit
+
+    def _resize(self, new_size: int) -> None:
+        old_buckets = self._buckets
+        self._buckets = [None] * new_size
+        self._len = 0
+        for item in old_buckets:
+            if item:
+                self._add_item(item.key, item.val)
+
+    def _size_up(self) -> None:
+        self._resize(len(self._buckets) * 2)
+
+    def _size_down(self) -> None:
+        self._resize(len(self._buckets) // 2)
+
+    def _iterate_buckets(self, key: KEY) -> Iterator[int]:
+        ind = self._get_bucket_index(key)
+        for _ in range(len(self._buckets)):
+            yield ind
+            ind = self._get_next_ind(ind)
+
+    def _add_item(self, key: KEY, val: VAL) -> None:
+        for ind in self._iterate_buckets(key):
+            if self._try_set(ind, key, val):
+                break
+
+    def __setitem__(self, key: KEY, val: VAL) -> None:
+        if self._is_full():
+            self._size_up()
+
+        self._add_item(key, val)
+
+    def __delitem__(self, key: KEY) -> None:
+        for ind in self._iterate_buckets(key):
+            item = self._buckets[ind]
+            if item is None:
+                raise KeyError(key)
+            if item is _deleted:
+                continue
+            if item.key == key:
+                self._buckets[ind] = _deleted
+                self._len -= 1
+                break
+        if self._is_sparse():
+            self._size_down()
+
+    def __getitem__(self, key: KEY) -> VAL:
+        for ind in self._iterate_buckets(key):
+            item = self._buckets[ind]
+            if item is None:
+                break
+            if item is _deleted:
+                continue
+            if item.key == key:
+                return item.val
+        raise KeyError(key)
+
+    def __len__(self) -> int:
+        return self._len
+
+    def __iter__(self) -> Iterator[KEY]:
+        yield from (item.key for item in self._buckets if item)
+
+    def __repr__(self) -> str:
+        val_string = " ,".join(
+            f"{item.key}: {item.val}" for item in self._buckets if item
+        )
+        return f"HashMap({val_string})"

data_structures/hashing/tests/test_hash_map.py

@@ -0,0 +1,97 @@
+from operator import delitem, getitem, setitem
+
+import pytest
+
+from data_structures.hashing.hash_map import HashMap
+
+
+def _get(k):
+    return getitem, k
+
+
+def _set(k, v):
+    return setitem, k, v
+
+
+def _del(k):
+    return delitem, k
+
+
+def _run_operation(obj, fun, *args):
+    try:
+        return fun(obj, *args), None
+    except Exception as e:
+        return None, e
+
+
+_add_items = (
+    _set("key_a", "val_a"),
+    _set("key_b", "val_b"),
+)
+
+_overwrite_items = [
+    _set("key_a", "val_a"),
+    _set("key_a", "val_b"),
+]
+
+_delete_items = [
+    _set("key_a", "val_a"),
+    _set("key_b", "val_b"),
+    _del("key_a"),
+    _del("key_b"),
+    _set("key_a", "val_a"),
+    _del("key_a"),
+]
+
+_access_absent_items = [
+    _get("key_a"),
+    _del("key_a"),
+    _set("key_a", "val_a"),
+    _del("key_a"),
+    _del("key_a"),
+    _get("key_a"),
+]
+
+_add_with_resize_up = [
+    *[_set(x, x) for x in range(5)],  # guaranteed upsize
+]
+
+_add_with_resize_down = [
+    *[_set(x, x) for x in range(5)],  # guaranteed upsize
+    *[_del(x) for x in range(5)],
+    _set("key_a", "val_b"),
+]
+
+
+@pytest.mark.parametrize(
+    "operations",
+    (
+        pytest.param(_add_items, id="add items"),
+        pytest.param(_overwrite_items, id="overwrite items"),
+        pytest.param(_delete_items, id="delete items"),
+        pytest.param(_access_absent_items, id="access absent items"),
+        pytest.param(_add_with_resize_up, id="add with resize up"),
+        pytest.param(_add_with_resize_down, id="add with resize down"),
+    ),
+)
+def test_hash_map_is_the_same_as_dict(operations):
+    my = HashMap(initial_block_size=4)
+    py = {}
+    for _, (fun, *args) in enumerate(operations):
+        my_res, my_exc = _run_operation(my, fun, *args)
+        py_res, py_exc = _run_operation(py, fun, *args)
+        assert my_res == py_res
+        assert str(my_exc) == str(py_exc)
+        assert set(py) == set(my)
+        assert len(py) == len(my)
+        assert set(my.items()) == set(py.items())
+
+
+def test_no_new_methods_was_added_to_api():
+    def is_public(name: str) -> bool:
+        return not name.startswith("_")
+
+    dict_public_names = {name for name in dir({}) if is_public(name)}
+    hash_public_names = {name for name in dir(HashMap()) if is_public(name)}
+
+    assert dict_public_names > hash_public_names