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

Determine whether an object is a SuccessTuple.