sqlglot.helper
1from __future__ import annotations 2 3import datetime 4import inspect 5import logging 6import re 7import sys 8import typing as t 9from collections.abc import Collection, Set, Iterable, Sequence, Iterator, Mapping 10from copy import copy 11from decimal import Decimal, InvalidOperation 12from difflib import get_close_matches 13from enum import Enum 14from itertools import count 15from builtins import type as Type 16 17try: 18 from mypy_extensions import mypyc_attr, trait, i64 19except ImportError: 20 21 def mypyc_attr(*attrs: str, **kwattrs: object) -> t.Callable[[t.Any], t.Any]: # type: ignore[misc] 22 return lambda f: f 23 24 def trait(f: t.Any) -> t.Any: # type: ignore[misc] 25 return f 26 27 i64 = int # type: ignore[misc,assignment] 28 29 30T = t.TypeVar("T") 31E = t.TypeVar("E") 32 33if t.TYPE_CHECKING: 34 from sqlglot.expressions import Expr 35 36 37CAMEL_CASE_PATTERN = re.compile("(?<!^)(?=[A-Z])") 38PYTHON_VERSION = sys.version_info[:2] 39logger = logging.getLogger("sqlglot") 40 41 42class AutoName(Enum): 43 """ 44 This is used for creating Enum classes where `auto()` is the string form 45 of the corresponding enum's identifier (e.g. FOO.value results in "FOO"). 46 47 Reference: https://docs.python.org/3/howto/enum.html#using-automatic-values 48 """ 49 50 def _generate_next_value_(name, _start, _count, _last_values): 51 return name 52 53 54def suggest_closest_match_and_fail( 55 kind: str, 56 word: str, 57 possibilities: Iterable[str], 58) -> None: 59 close_matches = get_close_matches(word, possibilities, n=1) 60 61 similar = seq_get(close_matches, 0) or "" 62 if similar: 63 similar = f" Did you mean {similar}?" 64 65 raise ValueError(f"Unknown {kind} '{word}'.{similar}") 66 67 68def seq_get(seq: Sequence[T], index: int) -> T | None: 69 """Returns the value in `seq` at position `index`, or `None` if `index` is out of bounds.""" 70 try: 71 return seq[index] 72 except IndexError: 73 return None 74 75 76def ensure_list(value: T | list[T] | tuple[T, ...] | None) -> list[T]: 77 """ 78 Ensures that a value is a list, otherwise casts or wraps it into one. 79 80 Args: 81 value: The value of interest. 82 83 Returns: 84 The value cast as a list if it's a list or a tuple, or else the value wrapped in a list. 85 """ 86 if value is None: 87 return [] 88 if isinstance(value, list): 89 return value 90 if isinstance(value, tuple): 91 return list(value) 92 93 return [value] 94 95 96@t.overload 97def ensure_collection(value: Collection[T]) -> Collection[T]: ... 98 99 100@t.overload 101def ensure_collection(value: T) -> Collection[T]: ... 102 103 104def ensure_collection(value): 105 """ 106 Ensures that a value is a collection (excluding `str` and `bytes`), otherwise wraps it into a list. 107 108 Args: 109 value: The value of interest. 110 111 Returns: 112 The value if it's a collection, or else the value wrapped in a list. 113 """ 114 if value is None: 115 return [] 116 return ( 117 value if isinstance(value, Collection) and not isinstance(value, (str, bytes)) else [value] 118 ) 119 120 121def csv(*args: str, sep: str = ", ") -> str: 122 """ 123 Formats any number of string arguments as CSV. 124 125 Args: 126 args: The string arguments to format. 127 sep: The argument separator. 128 129 Returns: 130 The arguments formatted as a CSV string. 131 """ 132 return sep.join(arg for arg in args if arg) 133 134 135def subclasses( 136 module_name: str, 137 classes: Type[T] | tuple[Type[T], ...], 138 exclude: set[Type[T]] = set(), 139) -> list[Type[T]]: 140 """ 141 Returns all subclasses for a collection of classes, possibly excluding some of them. 142 143 Args: 144 module_name: The name of the module to search for subclasses in. 145 classes: Class(es) we want to find the subclasses of. 146 exclude: Classes we want to exclude from the returned list. 147 148 Returns: 149 The target subclasses. 150 """ 151 return [ 152 obj 153 for _, obj in inspect.getmembers( 154 sys.modules[module_name], 155 lambda obj: inspect.isclass(obj) and issubclass(obj, classes) and obj not in exclude, 156 ) 157 ] 158 159 160def camel_to_snake_case(name: str) -> str: 161 """Converts `name` from camelCase to snake_case and returns the result.""" 162 return CAMEL_CASE_PATTERN.sub("_", name).upper() 163 164 165def while_changing(expression: E, func: t.Callable[[E], E]) -> E: 166 """ 167 Applies a transformation to a given expression until a fix point is reached. 168 169 Args: 170 expression: The expression to be transformed. 171 func: The transformation to be applied. 172 173 Returns: 174 The transformed expression. 175 """ 176 177 while True: 178 start_hash = hash(expression) 179 expression = func(expression) 180 end_hash = hash(expression) 181 182 if start_hash == end_hash: 183 break 184 185 return expression 186 187 188def tsort(dag: dict[T, set[T]]) -> list[T]: 189 """ 190 Sorts a given directed acyclic graph in topological order. 191 192 Args: 193 dag: The graph to be sorted. 194 195 Returns: 196 A list that contains all of the graph's nodes in topological order. 197 """ 198 result = [] 199 200 for node, deps in tuple(dag.items()): 201 for dep in deps: 202 if dep not in dag: 203 dag[dep] = set() 204 205 while dag: 206 current = {node for node, deps in dag.items() if not deps} 207 208 if not current: 209 raise ValueError("Cycle error") 210 211 for node in current: 212 dag.pop(node) 213 214 for deps in dag.values(): 215 deps -= current 216 217 result.extend(sorted(current)) # type: ignore 218 219 return result 220 221 222def find_new_name(taken: Collection[str], base: str) -> str: 223 """ 224 Searches for a new name. 225 226 Args: 227 taken: A collection of taken names. 228 base: Base name to alter. 229 230 Returns: 231 The new, available name. 232 """ 233 if base not in taken: 234 return base 235 236 i = 2 237 new = f"{base}_{i}" 238 while new in taken: 239 i += 1 240 new = f"{base}_{i}" 241 242 return new 243 244 245def is_int(text: str) -> bool: 246 return is_type(text, int) 247 248 249def to_int(value: str | int | Decimal, max_digits: int = 4300) -> int | None: 250 """Unlike int(), never materializes a huge int from a short exponent like 1e1000000.""" 251 try: 252 number = Decimal(value) 253 except InvalidOperation: 254 return None 255 256 if ( 257 not number.is_finite() 258 or number.adjusted() >= max_digits 259 or number != number.to_integral_value() 260 ): 261 return None 262 return int(number) 263 264 265def is_float(text: str) -> bool: 266 return is_type(text, float) 267 268 269def is_type(text: str, target_type: Type) -> bool: 270 try: 271 target_type(text) 272 return True 273 except ValueError: 274 return False 275 276 277def name_sequence(prefix: str) -> t.Callable[[], str]: 278 """Returns a name generator given a prefix (e.g. a0, a1, a2, ... if the prefix is "a").""" 279 sequence = count() 280 return lambda: f"{prefix}{next(sequence)}" 281 282 283def object_to_dict(obj: t.Any, **kwargs) -> dict: 284 """Returns a dictionary created from an object's attributes.""" 285 return { 286 **{k: v.copy() if hasattr(v, "copy") else copy(v) for k, v in vars(obj).items()}, 287 **kwargs, 288 } 289 290 291def split_num_words( 292 value: str, sep: str, min_num_words: int, fill_from_start: bool = True 293) -> list[str | None]: 294 """ 295 Perform a split on a value and return N words as a result with `None` used for words that don't exist. 296 297 Args: 298 value: The value to be split. 299 sep: The value to use to split on. 300 min_num_words: The minimum number of words that are going to be in the result. 301 fill_from_start: Indicates that if `None` values should be inserted at the start or end of the list. 302 303 Examples: 304 >>> split_num_words("db.table", ".", 3) 305 [None, 'db', 'table'] 306 >>> split_num_words("db.table", ".", 3, fill_from_start=False) 307 ['db', 'table', None] 308 >>> split_num_words("db.table", ".", 1) 309 ['db', 'table'] 310 311 Returns: 312 The list of words returned by `split`, possibly augmented by a number of `None` values. 313 """ 314 words = value.split(sep) 315 if fill_from_start: 316 return [None] * (min_num_words - len(words)) + words 317 return words + [None] * (min_num_words - len(words)) 318 319 320def is_iterable(value: t.Any) -> bool: 321 """ 322 Checks if the value is an iterable, excluding the types `str` and `bytes`. 323 324 Examples: 325 >>> is_iterable([1,2]) 326 True 327 >>> is_iterable("test") 328 False 329 330 Args: 331 value: The value to check if it is an iterable. 332 333 Returns: 334 A `bool` value indicating if it is an iterable. 335 """ 336 from sqlglot.expressions import Expr 337 338 return hasattr(value, "__iter__") and not isinstance(value, (str, bytes, Expr)) 339 340 341def flatten(values: Iterable[Iterable[t.Any] | t.Any]) -> Iterator[t.Any]: 342 """ 343 Flattens an iterable that can contain both iterable and non-iterable elements. Objects of 344 type `str` and `bytes` are not regarded as iterables. 345 346 Examples: 347 >>> list(flatten([[1, 2], 3, {4}, (5, "bla")])) 348 [1, 2, 3, 4, 5, 'bla'] 349 >>> list(flatten([1, 2, 3])) 350 [1, 2, 3] 351 352 Args: 353 values: The value to be flattened. 354 355 Yields: 356 Non-iterable elements in `values`. 357 """ 358 for value in values: 359 if is_iterable(value): 360 yield from flatten(value) 361 else: 362 yield value 363 364 365def dict_depth(d: t.Any) -> int: 366 """ 367 Get the nesting depth of a dictionary. 368 369 Example: 370 >>> dict_depth(None) 371 0 372 >>> dict_depth({}) 373 1 374 >>> dict_depth({"a": "b"}) 375 1 376 >>> dict_depth({"a": {}}) 377 2 378 >>> dict_depth({"a": {"b": {}}}) 379 3 380 """ 381 try: 382 return 1 + dict_depth(next(iter(d.values()))) 383 except AttributeError: 384 # d doesn't have attribute "values" 385 return 0 386 except StopIteration: 387 # d.values() returns an empty sequence 388 return 1 389 390 391def first(it: Iterable[T]) -> T: 392 """Returns the first element from an iterable (useful for sets).""" 393 return next(i for i in it) 394 395 396@t.overload 397def to_bool(value: None) -> None: ... 398 399 400@t.overload 401def to_bool(value: bool) -> bool: ... 402 403 404@t.overload 405def to_bool(value: str) -> str | bool: ... 406 407 408def to_bool(value: str | bool | None) -> str | bool | None: 409 if isinstance(value, bool) or value is None: 410 return value 411 412 # Coerce the value to boolean if it matches to the truthy/falsy values below 413 value_lower = value.lower() 414 if value_lower in ("true", "1"): 415 return True 416 if value_lower in ("false", "0"): 417 return False 418 419 return value 420 421 422def merge_ranges(ranges: list[tuple[t.Any, t.Any]]) -> list[tuple[t.Any, t.Any]]: 423 """ 424 Merges a sequence of ranges, represented as tuples (low, high) whose values 425 belong to some totally-ordered set. 426 427 Example: 428 >>> merge_ranges([(1, 3), (2, 6)]) 429 [(1, 6)] 430 """ 431 if not ranges: 432 return [] 433 434 ranges = sorted(ranges) 435 436 merged = [ranges[0]] 437 438 for start, end in ranges[1:]: 439 last_start, last_end = merged[-1] 440 441 if start <= last_end: 442 merged[-1] = (last_start, max(last_end, end)) 443 else: 444 merged.append((start, end)) 445 446 return merged 447 448 449def is_iso_date(text: str) -> bool: 450 try: 451 datetime.date.fromisoformat(text) 452 return True 453 except ValueError: 454 return False 455 456 457def is_iso_datetime(text: str) -> bool: 458 try: 459 datetime.datetime.fromisoformat(text) 460 return True 461 except ValueError: 462 return False 463 464 465# Interval units that operate on date components 466DATE_UNITS = {"day", "week", "month", "quarter", "year", "year_month"} 467 468 469def is_date_unit(expression: Expr | None) -> bool: 470 return expression is not None and expression.name.lower() in DATE_UNITS 471 472 473K = t.TypeVar("K") 474V = t.TypeVar("V") 475 476 477class SingleValuedMapping(Mapping[K, V]): 478 """ 479 Mapping where all keys return the same value. 480 481 This rigamarole is meant to avoid copying keys, which was originally intended 482 as an optimization while qualifying columns for tables with lots of columns. 483 """ 484 485 def __init__(self, keys: Collection[K], value: V): 486 self._keys = keys if isinstance(keys, Set) else set(keys) 487 self._value = value 488 489 def __getitem__(self, key: K) -> V: 490 if key in self._keys: 491 return self._value 492 raise KeyError(key) 493 494 def __len__(self) -> int: 495 return len(self._keys) 496 497 def __iter__(self) -> Iterator[K]: 498 return iter(self._keys)
CAMEL_CASE_PATTERN =
re.compile('(?<!^)(?=[A-Z])')
PYTHON_VERSION =
(3, 10)
logger =
<Logger sqlglot (WARNING)>
class
AutoName(enum.Enum):
43class AutoName(Enum): 44 """ 45 This is used for creating Enum classes where `auto()` is the string form 46 of the corresponding enum's identifier (e.g. FOO.value results in "FOO"). 47 48 Reference: https://docs.python.org/3/howto/enum.html#using-automatic-values 49 """ 50 51 def _generate_next_value_(name, _start, _count, _last_values): 52 return name
This is used for creating Enum classes where auto() is the string form
of the corresponding enum's identifier (e.g. FOO.value results in "FOO").
Reference: https://docs.python.org/3/howto/enum.html#using-automatic-values
def
suggest_closest_match_and_fail(kind: str, word: str, possibilities: Iterable[str]) -> None:
55def suggest_closest_match_and_fail( 56 kind: str, 57 word: str, 58 possibilities: Iterable[str], 59) -> None: 60 close_matches = get_close_matches(word, possibilities, n=1) 61 62 similar = seq_get(close_matches, 0) or "" 63 if similar: 64 similar = f" Did you mean {similar}?" 65 66 raise ValueError(f"Unknown {kind} '{word}'.{similar}")
def
seq_get(seq: Sequence[~T], index: int) -> Optional[~T]:
69def seq_get(seq: Sequence[T], index: int) -> T | None: 70 """Returns the value in `seq` at position `index`, or `None` if `index` is out of bounds.""" 71 try: 72 return seq[index] 73 except IndexError: 74 return None
Returns the value in seq at position index, or None if index is out of bounds.
def
ensure_list(value: Union[~T, list[~T], tuple[~T, ...], NoneType]) -> list[~T]:
77def ensure_list(value: T | list[T] | tuple[T, ...] | None) -> list[T]: 78 """ 79 Ensures that a value is a list, otherwise casts or wraps it into one. 80 81 Args: 82 value: The value of interest. 83 84 Returns: 85 The value cast as a list if it's a list or a tuple, or else the value wrapped in a list. 86 """ 87 if value is None: 88 return [] 89 if isinstance(value, list): 90 return value 91 if isinstance(value, tuple): 92 return list(value) 93 94 return [value]
Ensures that a value is a list, otherwise casts or wraps it into one.
Arguments:
- value: The value of interest.
Returns:
The value cast as a list if it's a list or a tuple, or else the value wrapped in a list.
def
ensure_collection(value):
105def ensure_collection(value): 106 """ 107 Ensures that a value is a collection (excluding `str` and `bytes`), otherwise wraps it into a list. 108 109 Args: 110 value: The value of interest. 111 112 Returns: 113 The value if it's a collection, or else the value wrapped in a list. 114 """ 115 if value is None: 116 return [] 117 return ( 118 value if isinstance(value, Collection) and not isinstance(value, (str, bytes)) else [value] 119 )
Ensures that a value is a collection (excluding str and bytes), otherwise wraps it into a list.
Arguments:
- value: The value of interest.
Returns:
The value if it's a collection, or else the value wrapped in a list.
def
csv(*args: str, sep: str = ', ') -> str:
122def csv(*args: str, sep: str = ", ") -> str: 123 """ 124 Formats any number of string arguments as CSV. 125 126 Args: 127 args: The string arguments to format. 128 sep: The argument separator. 129 130 Returns: 131 The arguments formatted as a CSV string. 132 """ 133 return sep.join(arg for arg in args if arg)
Formats any number of string arguments as CSV.
Arguments:
- args: The string arguments to format.
- sep: The argument separator.
Returns:
The arguments formatted as a CSV string.
def
subclasses( module_name: str, classes: type[~T] | tuple[type[~T], ...], exclude: set[type[~T]] = set()) -> list[type[~T]]:
136def subclasses( 137 module_name: str, 138 classes: Type[T] | tuple[Type[T], ...], 139 exclude: set[Type[T]] = set(), 140) -> list[Type[T]]: 141 """ 142 Returns all subclasses for a collection of classes, possibly excluding some of them. 143 144 Args: 145 module_name: The name of the module to search for subclasses in. 146 classes: Class(es) we want to find the subclasses of. 147 exclude: Classes we want to exclude from the returned list. 148 149 Returns: 150 The target subclasses. 151 """ 152 return [ 153 obj 154 for _, obj in inspect.getmembers( 155 sys.modules[module_name], 156 lambda obj: inspect.isclass(obj) and issubclass(obj, classes) and obj not in exclude, 157 ) 158 ]
Returns all subclasses for a collection of classes, possibly excluding some of them.
Arguments:
- module_name: The name of the module to search for subclasses in.
- classes: Class(es) we want to find the subclasses of.
- exclude: Classes we want to exclude from the returned list.
Returns:
The target subclasses.
def
camel_to_snake_case(name: str) -> str:
161def camel_to_snake_case(name: str) -> str: 162 """Converts `name` from camelCase to snake_case and returns the result.""" 163 return CAMEL_CASE_PATTERN.sub("_", name).upper()
Converts name from camelCase to snake_case and returns the result.
def
while_changing(expression: ~E, func: Callable[[~E], ~E]) -> ~E:
166def while_changing(expression: E, func: t.Callable[[E], E]) -> E: 167 """ 168 Applies a transformation to a given expression until a fix point is reached. 169 170 Args: 171 expression: The expression to be transformed. 172 func: The transformation to be applied. 173 174 Returns: 175 The transformed expression. 176 """ 177 178 while True: 179 start_hash = hash(expression) 180 expression = func(expression) 181 end_hash = hash(expression) 182 183 if start_hash == end_hash: 184 break 185 186 return expression
Applies a transformation to a given expression until a fix point is reached.
Arguments:
- expression: The expression to be transformed.
- func: The transformation to be applied.
Returns:
The transformed expression.
def
tsort(dag: dict[~T, set[~T]]) -> list[~T]:
189def tsort(dag: dict[T, set[T]]) -> list[T]: 190 """ 191 Sorts a given directed acyclic graph in topological order. 192 193 Args: 194 dag: The graph to be sorted. 195 196 Returns: 197 A list that contains all of the graph's nodes in topological order. 198 """ 199 result = [] 200 201 for node, deps in tuple(dag.items()): 202 for dep in deps: 203 if dep not in dag: 204 dag[dep] = set() 205 206 while dag: 207 current = {node for node, deps in dag.items() if not deps} 208 209 if not current: 210 raise ValueError("Cycle error") 211 212 for node in current: 213 dag.pop(node) 214 215 for deps in dag.values(): 216 deps -= current 217 218 result.extend(sorted(current)) # type: ignore 219 220 return result
Sorts a given directed acyclic graph in topological order.
Arguments:
- dag: The graph to be sorted.
Returns:
A list that contains all of the graph's nodes in topological order.
def
find_new_name(taken: Collection[str], base: str) -> str:
223def find_new_name(taken: Collection[str], base: str) -> str: 224 """ 225 Searches for a new name. 226 227 Args: 228 taken: A collection of taken names. 229 base: Base name to alter. 230 231 Returns: 232 The new, available name. 233 """ 234 if base not in taken: 235 return base 236 237 i = 2 238 new = f"{base}_{i}" 239 while new in taken: 240 i += 1 241 new = f"{base}_{i}" 242 243 return new
Searches for a new name.
Arguments:
- taken: A collection of taken names.
- base: Base name to alter.
Returns:
The new, available name.
def
is_int(text: str) -> bool:
def
to_int(value: str | int | decimal.Decimal, max_digits: int = 4300) -> int | None:
250def to_int(value: str | int | Decimal, max_digits: int = 4300) -> int | None: 251 """Unlike int(), never materializes a huge int from a short exponent like 1e1000000.""" 252 try: 253 number = Decimal(value) 254 except InvalidOperation: 255 return None 256 257 if ( 258 not number.is_finite() 259 or number.adjusted() >= max_digits 260 or number != number.to_integral_value() 261 ): 262 return None 263 return int(number)
Unlike int(), never materializes a huge int from a short exponent like 1e1000000.
def
is_float(text: str) -> bool:
def
is_type(text: str, target_type: type) -> bool:
def
name_sequence(prefix: str) -> Callable[[], str]:
278def name_sequence(prefix: str) -> t.Callable[[], str]: 279 """Returns a name generator given a prefix (e.g. a0, a1, a2, ... if the prefix is "a").""" 280 sequence = count() 281 return lambda: f"{prefix}{next(sequence)}"
Returns a name generator given a prefix (e.g. a0, a1, a2, ... if the prefix is "a").
def
object_to_dict(obj: Any, **kwargs) -> dict:
284def object_to_dict(obj: t.Any, **kwargs) -> dict: 285 """Returns a dictionary created from an object's attributes.""" 286 return { 287 **{k: v.copy() if hasattr(v, "copy") else copy(v) for k, v in vars(obj).items()}, 288 **kwargs, 289 }
Returns a dictionary created from an object's attributes.
def
split_num_words( value: str, sep: str, min_num_words: int, fill_from_start: bool = True) -> list[str | None]:
292def split_num_words( 293 value: str, sep: str, min_num_words: int, fill_from_start: bool = True 294) -> list[str | None]: 295 """ 296 Perform a split on a value and return N words as a result with `None` used for words that don't exist. 297 298 Args: 299 value: The value to be split. 300 sep: The value to use to split on. 301 min_num_words: The minimum number of words that are going to be in the result. 302 fill_from_start: Indicates that if `None` values should be inserted at the start or end of the list. 303 304 Examples: 305 >>> split_num_words("db.table", ".", 3) 306 [None, 'db', 'table'] 307 >>> split_num_words("db.table", ".", 3, fill_from_start=False) 308 ['db', 'table', None] 309 >>> split_num_words("db.table", ".", 1) 310 ['db', 'table'] 311 312 Returns: 313 The list of words returned by `split`, possibly augmented by a number of `None` values. 314 """ 315 words = value.split(sep) 316 if fill_from_start: 317 return [None] * (min_num_words - len(words)) + words 318 return words + [None] * (min_num_words - len(words))
Perform a split on a value and return N words as a result with None used for words that don't exist.
Arguments:
- value: The value to be split.
- sep: The value to use to split on.
- min_num_words: The minimum number of words that are going to be in the result.
- fill_from_start: Indicates that if
Nonevalues should be inserted at the start or end of the list.
Examples:
>>> split_num_words("db.table", ".", 3) [None, 'db', 'table'] >>> split_num_words("db.table", ".", 3, fill_from_start=False) ['db', 'table', None] >>> split_num_words("db.table", ".", 1) ['db', 'table']
Returns:
The list of words returned by
split, possibly augmented by a number ofNonevalues.
def
is_iterable(value: Any) -> bool:
321def is_iterable(value: t.Any) -> bool: 322 """ 323 Checks if the value is an iterable, excluding the types `str` and `bytes`. 324 325 Examples: 326 >>> is_iterable([1,2]) 327 True 328 >>> is_iterable("test") 329 False 330 331 Args: 332 value: The value to check if it is an iterable. 333 334 Returns: 335 A `bool` value indicating if it is an iterable. 336 """ 337 from sqlglot.expressions import Expr 338 339 return hasattr(value, "__iter__") and not isinstance(value, (str, bytes, Expr))
Checks if the value is an iterable, excluding the types str and bytes.
Examples:
>>> is_iterable([1,2]) True >>> is_iterable("test") False
Arguments:
- value: The value to check if it is an iterable.
Returns:
A
boolvalue indicating if it is an iterable.
def
flatten( values: Iterable[typing.Union[Iterable[typing.Any], typing.Any]]) -> Iterator[typing.Any]:
342def flatten(values: Iterable[Iterable[t.Any] | t.Any]) -> Iterator[t.Any]: 343 """ 344 Flattens an iterable that can contain both iterable and non-iterable elements. Objects of 345 type `str` and `bytes` are not regarded as iterables. 346 347 Examples: 348 >>> list(flatten([[1, 2], 3, {4}, (5, "bla")])) 349 [1, 2, 3, 4, 5, 'bla'] 350 >>> list(flatten([1, 2, 3])) 351 [1, 2, 3] 352 353 Args: 354 values: The value to be flattened. 355 356 Yields: 357 Non-iterable elements in `values`. 358 """ 359 for value in values: 360 if is_iterable(value): 361 yield from flatten(value) 362 else: 363 yield value
Flattens an iterable that can contain both iterable and non-iterable elements. Objects of
type str and bytes are not regarded as iterables.
Examples:
>>> list(flatten([[1, 2], 3, {4}, (5, "bla")])) [1, 2, 3, 4, 5, 'bla'] >>> list(flatten([1, 2, 3])) [1, 2, 3]
Arguments:
- values: The value to be flattened.
Yields:
Non-iterable elements in
values.
def
dict_depth(d: Any) -> int:
366def dict_depth(d: t.Any) -> int: 367 """ 368 Get the nesting depth of a dictionary. 369 370 Example: 371 >>> dict_depth(None) 372 0 373 >>> dict_depth({}) 374 1 375 >>> dict_depth({"a": "b"}) 376 1 377 >>> dict_depth({"a": {}}) 378 2 379 >>> dict_depth({"a": {"b": {}}}) 380 3 381 """ 382 try: 383 return 1 + dict_depth(next(iter(d.values()))) 384 except AttributeError: 385 # d doesn't have attribute "values" 386 return 0 387 except StopIteration: 388 # d.values() returns an empty sequence 389 return 1
Get the nesting depth of a dictionary.
Example:
>>> dict_depth(None) 0 >>> dict_depth({}) 1 >>> dict_depth({"a": "b"}) 1 >>> dict_depth({"a": {}}) 2 >>> dict_depth({"a": {"b": {}}}) 3
def
first(it: Iterable[~T]) -> ~T:
392def first(it: Iterable[T]) -> T: 393 """Returns the first element from an iterable (useful for sets).""" 394 return next(i for i in it)
Returns the first element from an iterable (useful for sets).
def
to_bool(value: str | bool | None) -> str | bool | None:
409def to_bool(value: str | bool | None) -> str | bool | None: 410 if isinstance(value, bool) or value is None: 411 return value 412 413 # Coerce the value to boolean if it matches to the truthy/falsy values below 414 value_lower = value.lower() 415 if value_lower in ("true", "1"): 416 return True 417 if value_lower in ("false", "0"): 418 return False 419 420 return value
def
merge_ranges( ranges: list[tuple[typing.Any, typing.Any]]) -> list[tuple[typing.Any, typing.Any]]:
423def merge_ranges(ranges: list[tuple[t.Any, t.Any]]) -> list[tuple[t.Any, t.Any]]: 424 """ 425 Merges a sequence of ranges, represented as tuples (low, high) whose values 426 belong to some totally-ordered set. 427 428 Example: 429 >>> merge_ranges([(1, 3), (2, 6)]) 430 [(1, 6)] 431 """ 432 if not ranges: 433 return [] 434 435 ranges = sorted(ranges) 436 437 merged = [ranges[0]] 438 439 for start, end in ranges[1:]: 440 last_start, last_end = merged[-1] 441 442 if start <= last_end: 443 merged[-1] = (last_start, max(last_end, end)) 444 else: 445 merged.append((start, end)) 446 447 return merged
Merges a sequence of ranges, represented as tuples (low, high) whose values belong to some totally-ordered set.
Example:
>>> merge_ranges([(1, 3), (2, 6)]) [(1, 6)]
def
is_iso_date(text: str) -> bool:
def
is_iso_datetime(text: str) -> bool:
DATE_UNITS =
{'year_month', 'day', 'week', 'year', 'month', 'quarter'}
class
SingleValuedMapping(collections.abc.Mapping[~K, ~V]):
478class SingleValuedMapping(Mapping[K, V]): 479 """ 480 Mapping where all keys return the same value. 481 482 This rigamarole is meant to avoid copying keys, which was originally intended 483 as an optimization while qualifying columns for tables with lots of columns. 484 """ 485 486 def __init__(self, keys: Collection[K], value: V): 487 self._keys = keys if isinstance(keys, Set) else set(keys) 488 self._value = value 489 490 def __getitem__(self, key: K) -> V: 491 if key in self._keys: 492 return self._value 493 raise KeyError(key) 494 495 def __len__(self) -> int: 496 return len(self._keys) 497 498 def __iter__(self) -> Iterator[K]: 499 return iter(self._keys)
Mapping where all keys return the same value.
This rigamarole is meant to avoid copying keys, which was originally intended as an optimization while qualifying columns for tables with lots of columns.