meerschaum.utils.typing

  1from __future__ import annotations
  2try:
  3    from typing import (
  4        Tuple,
  5        Optional,
  6        Dict,
  7        List,
  8        Mapping,
  9        Sequence,
 10        Callable,
 11        Union,
 12        Any,
 13        Iterable,
 14        Hashable,
 15        Generator,
 16        Iterator,
 17        TYPE_CHECKING,
 18    )
 19except Exception as e:
 20    import urllib.request, sys, pathlib, os
 21    old_cwd = os.getcwd()
 22    cache_dir = pathlib.Path.home() / '.cache'
 23    if not cache_dir.exists():
 24        try:
 25            cache_dir.mkdir(parents=True, exist_ok=True)
 26        except Exception as _e:
 27            cache_dir = pathlib.Path.home()
 28
 29    dest_file = cache_dir / 'typing_hotfix.py'
 30    os.chdir(cache_dir)
 31
 32    url = 'https://raw.githubusercontent.com/python/typing_extensions/main/src/typing_extensions.py'
 33    if not dest_file.exists():
 34        response = urllib.request.urlopen(url)
 35        if response.code != 200:
 36            print(f"Could not download typing. Please install typing via pip or upgrade Python.")
 37            sys.exit(1)
 38        with open(dest_file, 'wb') as f:
 39            f.write(response.fp.read())
 40    
 41    import typing_hotfix
 42    os.chdir(old_cwd)
 43
 44### Patch Literal for Python 3.7.
 45try:
 46    from typing import Literal
 47except ImportError:
 48    import typing
 49
 50    class _LiteralForm(typing._SpecialForm, _root=True):
 51
 52        def __repr__(self):
 53            return 'typing_extensions.' + self._name
 54
 55        def __getitem__(self, parameters):
 56            return typing._GenericAlias(self, parameters)
 57
 58    typing.Literal = _LiteralForm(
 59        'Literal',
 60       doc = """A type that can be used to indicate to type checkers
 61       that the corresponding value has a value literally equivalent
 62       to the provided parameter. For example:
 63
 64           var: Literal[4] = 4
 65
 66       The type checker understands that 'var' is literally equal to
 67       the value 4 and no other value.
 68
 69       Literal[...] cannot be subclassed. There is no runtime
 70       checking verifying that the parameter is actually a value
 71       instead of a type."""
 72    )
 73
 74
 75import collections.abc
 76collections.Iterable = collections.abc.Iterable
 77
 78SuccessTuple = Tuple[bool, str]
 79InstanceConnector = Union[
 80    'meerschaum.connectors.sql.SQLConnector',
 81    'meerschaum.connectors.api.APIConnector'
 82]
 83PipesDict = Dict[
 84    str, Dict[                           ### connector_keys : metrics
 85        str, Dict[                       ### metric_key     : locations
 86            str, 'meerschaum.Pipe'       ### location_key   : Pipe
 87        ]
 88    ]
 89]
 90WebState = Dict[str, Union[str, Dict[str, str]]]
 91
 92def is_success_tuple(x: Any) -> bool:
 93    """
 94    Determine whether an object is a `SuccessTuple`.
 95    """
 96    return (
 97        isinstance(x, tuple)
 98        and len(x) == 2
 99        and isinstance(x[0], bool)
100        and isinstance(x[1], str)
101    )
SuccessTuple = typing.Tuple[bool, str]
InstanceConnector = typing.Union[ForwardRef('meerschaum.connectors.SQLConnector'), ForwardRef('meerschaum.connectors.APIConnector')]
PipesDict = typing.Dict[str, typing.Dict[str, typing.Dict[str, ForwardRef('meerschaum.Pipe')]]]
WebState = typing.Dict[str, typing.Union[str, typing.Dict[str, str]]]
def is_success_tuple(x: Any) -> bool:
 93def is_success_tuple(x: Any) -> bool:
 94    """
 95    Determine whether an object is a `SuccessTuple`.
 96    """
 97    return (
 98        isinstance(x, tuple)
 99        and len(x) == 2
100        and isinstance(x[0], bool)
101        and isinstance(x[1], str)
102    )

Determine whether an object is a SuccessTuple.