Source code for azure.core.utils._utils

# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
import datetime
import sys
from typing import (
from datetime import timezone

TZ_UTC = timezone.utc

class _FixedOffset(datetime.tzinfo):
    """Fixed offset in minutes east from UTC.

    Copy/pasted from Python doc

    :param int offset: offset in minutes

    def __init__(self, offset):
        self.__offset = datetime.timedelta(minutes=offset)

    def utcoffset(self, dt):
        return self.__offset

    def tzname(self, dt):
        return str(self.__offset.total_seconds() / 3600)

    def __repr__(self):
        return "<FixedOffset {}>".format(self.tzname(None))

    def dst(self, dt):
        return datetime.timedelta(0)

def _convert_to_isoformat(date_time):
    """Deserialize a date in RFC 3339 format to datetime object.
    Check for examples.

    :param str date_time: The date in RFC 3339 format.
    if not date_time:
        return None
    if date_time[-1] == "Z":
        delta = 0
        timestamp = date_time[:-1]
        timestamp = date_time[:-6]
        sign, offset = date_time[-6], date_time[-5:]
        delta = int(sign + offset[:1]) * 60 + int(sign + offset[-2:])

    check_decimal = timestamp.split(".")
    if len(check_decimal) > 1:
        decimal_str = ""
        for digit in check_decimal[1]:
            if digit.isdigit():
                decimal_str += digit
        if len(decimal_str) > 6:
            timestamp = timestamp.replace(decimal_str, decimal_str[0:6])

    if delta == 0:
        tzinfo = TZ_UTC
        tzinfo = timezone(datetime.timedelta(minutes=delta))

        deserialized = datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%f")
    except ValueError:
        deserialized = datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S")

    deserialized = deserialized.replace(tzinfo=tzinfo)
    return deserialized

[docs]def case_insensitive_dict( *args: Optional[Union[Mapping[str, Any], Iterable[Tuple[str, Any]]]], **kwargs: Any ) -> MutableMapping[str, Any]: """Return a case-insensitive mutable mapping from an inputted mapping structure. :param args: The positional arguments to pass to the dict. :type args: Mapping[str, Any] or Iterable[Tuple[str, Any] :return: A case-insensitive mutable mapping object. :rtype: """ return CaseInsensitiveDict(*args, **kwargs)
[docs]class CaseInsensitiveDict(MutableMapping[str, Any]): """ 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 :param data: Initial data to store in the dictionary. :type data: Mapping[str, Any] or Iterable[Tuple[str, Any]] """ def __init__( self, data: Optional[Union[Mapping[str, Any], Iterable[Tuple[str, Any]]]] = None, **kwargs: Any ) -> None: self._store: Dict[str, Any] = {} if data is None: data = {} self.update(data, **kwargs)
[docs] def copy(self) -> "CaseInsensitiveDict": return CaseInsensitiveDict(self._store.values())
def __setitem__(self, key: str, value: Any) -> None: """Set the `key` to `value`. The original key will be stored with the value :param str key: The key to set. :param value: The value to set the key to. :type value: any """ 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[str]: return (key for key, _ in self._store.values()) def __len__(self) -> int: return len(self._store)
[docs] def lowerkey_items(self) -> Iterator[Tuple[str, Any]]: 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()))
def get_running_async_lock() -> AsyncContextManager: """Get a lock instance from the async library that the current context is running under. :return: An instance of the running async library's Lock class. :rtype: AsyncContextManager :raises: RuntimeError if the current context is not running under an async library. """ try: import asyncio # Check if we are running in an asyncio event loop. asyncio.get_running_loop() return asyncio.Lock() except RuntimeError as err: # Otherwise, assume we are running in a trio event loop if it has already been imported. if "trio" in sys.modules: import trio # pylint: disable=networking-import-outside-azure-core-transport return trio.Lock() raise RuntimeError("An asyncio or trio event loop is required.") from err