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 = 'meerschaum.connectors.InstanceConnector'
80PipesDict = Dict[
81    str, Dict[                           ### connector_keys : metrics
82        str, Dict[                       ### metric_key     : locations
83            str, 'meerschaum.Pipe'       ### location_key   : Pipe
84        ]
85    ]
86]
87WebState = Dict[str, Union[str, Dict[str, str]]]
88
89def is_success_tuple(x: Any) -> bool:
90    """
91    Determine whether an object is a `SuccessTuple`.
92    """
93    return (
94        isinstance(x, tuple)
95        and len(x) == 2
96        and isinstance(x[0], bool)
97        and isinstance(x[1], str)
98    )
SuccessTuple = typing.Tuple[bool, str]
InstanceConnector = 'meerschaum.InstanceConnector'
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:
90def is_success_tuple(x: Any) -> bool:
91    """
92    Determine whether an object is a `SuccessTuple`.
93    """
94    return (
95        isinstance(x, tuple)
96        and len(x) == 2
97        and isinstance(x[0], bool)
98        and isinstance(x[1], str)
99    )

Determine whether an object is a SuccessTuple.