[Core] Case Insensitive Dictionary Implementation (Azure#25074)

* cidict implementation

* unit tests

* expose CaseInsensitiveDict

* changelog

* fix typing issues

* fix _iter_ mypy issue

* pylint fix

Author: kashifkhan

sdk/core/azure-core/CHANGELOG.md

@@ -6,6 +6,8 @@ Azure-core is supported on Python 3.7 or later. For more details, please read ou
 
 ### Features Added
 
+- Added `CaseInsensitiveDict` implementation in `azure.core.utils` removing dependency on `requests` and `aiohttp`
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/azure-core/azure/core/utils/__init__.py

@@ -32,6 +32,6 @@
 from ._connection_string_parser import (
     parse_connection_string
 )
-from ._utils import case_insensitive_dict
+from ._utils import case_insensitive_dict, CaseInsensitiveDict
 
-__all__ = ["parse_connection_string", "case_insensitive_dict"]
+__all__ = ["parse_connection_string", "case_insensitive_dict", "CaseInsensitiveDict"]

sdk/core/azure-core/azure/core/utils/_utils.py

@@ -5,7 +5,7 @@
 # license information.
 # --------------------------------------------------------------------------
 import datetime
-from typing import Any, MutableMapping
+from typing import Any, Dict, Iterator, Mapping, MutableMapping
 
 
 class _FixedOffset(datetime.tzinfo):
@@ -87,26 +87,59 @@ def case_insensitive_dict(*args: Any, **kwargs: Any) -> MutableMapping:
     :return: A case-insensitive mutable mapping object.
     :rtype: ~collections.abc.MutableMapping
     """
+    return CaseInsensitiveDict(*args, **kwargs)
 
-    # Rational is I don't want to re-implement this, but I don't want
-    # to assume "requests" or "aiohttp" are installed either.
-    # So I use the one from "requests" or the one from "aiohttp" ("multidict")
-    # If one day this library is used in an HTTP context without "requests" nor "aiohttp" installed,
-    # we can add "multidict" as a dependency or re-implement our own.
-    try:
-        from requests.structures import CaseInsensitiveDict
+class CaseInsensitiveDict(MutableMapping):
+    """
+    NOTE: This implementation is heavily inspired from the case insensitive dictionary from the requests library.
+    Thank you !!
+    Case insensitive dictionary implementation.
+    The keys are expected to be strings and will be stored in lower case.
+    case_insensitive_dict = CaseInsensitiveDict()
+    case_insensitive_dict['Key'] = 'some_value'
+    case_insensitive_dict['key'] == 'some_value' #True
+    """
 
-        return CaseInsensitiveDict(*args, **kwargs)
-    except ImportError:
-        pass
-    try:
-        # multidict is installed by aiohttp
-        from multidict import CIMultiDict
-
-        if len(kwargs) == 0 and len(args) == 1 and (not args[0]):
-            return CIMultiDict()    # in case of case_insensitive_dict(None), we don't want to raise exception
-        return CIMultiDict(*args, **kwargs)
-    except ImportError:
-        raise ValueError(
-            "Neither 'requests' or 'multidict' are installed and no case-insensitive dict impl have been found"
+    def __init__(self, data=None, **kwargs: Any) -> None:
+        self._store: Dict[str, Any] = {}
+        if data is None:
+            data = {}
+
+        self.update(data, **kwargs)
+
+    def copy(self) -> "CaseInsensitiveDict":
+        return CaseInsensitiveDict(self._store.values())
+
+    def __setitem__(self, key: str, value: str) -> None:
+        """
+        Set the `key` to `value`. The original key will be stored with the value
+        """
+        self._store[key.lower()] = (key, value)
+
+    def __getitem__(self, key: str) -> Any:
+        return self._store[key.lower()][1]
+
+    def __delitem__(self, key: str) -> None:
+        del self._store[key.lower()]
+
+    def __iter__(self) -> Iterator[Any]:
+        return (key for key, _ in self._store.values())
+
+    def __len__(self) -> int:
+        return len(self._store)
+
+    def lowerkey_items(self):
+        return (
+            (lower_case_key, pair[1]) for lower_case_key, pair in self._store.items()
         )
+
+    def __eq__(self, other: Any) -> bool:
+        if isinstance(other, Mapping):
+            other = CaseInsensitiveDict(other)
+        else:
+            return False
+
+        return dict(self.lowerkey_items()) == dict(other.lowerkey_items())
+
+    def __repr__(self) -> str:
+        return str(dict(self.items()))

sdk/core/azure-core/tests/test_utils.py

@@ -40,3 +40,56 @@ def test_case_insensitive_dict_initialization():
         assert d['platformUpdateDomainCount'] == d['platformupdatedomaincount'] == d['PLATFORMUPDATEDOMAINCOUNT'] == 5
         assert d['platformFaultDomainCount'] == d['platformfaultdomaincount'] == d['PLATFORMFAULTDOMAINCOUNT'] == 3
         assert d['virtualMachines'] == d['virtualmachines'] == d['VIRTUALMACHINES'] == []
+
+def test_case_insensitive_dict_cant_compare():
+    my_dict = case_insensitive_dict({"accept": "application/json"})
+    assert my_dict != "accept"
+
+def test_case_insensitive_dict_lowerkey_items():
+    my_dict = case_insensitive_dict({"accept": "application/json"})
+    assert list(my_dict.lowerkey_items()) == [("accept","application/json")]
+
+@pytest.mark.parametrize("other, expected", (
+    ({"PLATFORMUPDATEDOMAINCOUNT": 5}, True),
+    ({}, False),
+    (None, False),
+))
+def test_case_insensitive_dict_equality(other, expected):
+    my_dict = case_insensitive_dict({"platformUpdateDomainCount": 5})
+    result = my_dict == other
+    assert result == expected
+
+def test_case_insensitive_dict_keys():
+    keys = ["One", "TWO", "tHrEe", "four"]
+    my_dict = case_insensitive_dict({key:idx for idx, key in enumerate(keys,1)})
+    dict_keys = list(my_dict.keys())
+
+    assert dict_keys == keys
+
+def test_case_insensitive_copy():
+    keys = ["One", "TWO", "tHrEe", "four"]
+    my_dict = case_insensitive_dict({key:idx for idx, key in enumerate(keys, 1)})
+    copy_my_dict = my_dict.copy()
+    assert copy_my_dict is not my_dict
+    assert copy_my_dict == my_dict
+
+def test_case_insensitive_keys_present(accept_cases):
+    my_dict = case_insensitive_dict({"accept": "application/json"})
+
+    for key in accept_cases:
+        assert key in my_dict
+
+def test_case_insensitive_keys_delete(accept_cases):
+    my_dict = case_insensitive_dict({"accept": "application/json"})
+
+    for key in accept_cases:
+        del my_dict[key]
+        assert key not in my_dict
+        my_dict[key] = "application/json"
+
+def test_case_iter():
+    keys = ["One", "TWO", "tHrEe", "four"]
+    my_dict = case_insensitive_dict({key:idx for idx, key in enumerate(keys, 1)})
+    
+    for key in my_dict:
+        assert key in keys