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
.