Path: blob/main/singlestoredb/utils/convert_rows.py
469 views
#!/usr/bin/env python1"""Data value conversion utilities."""2from typing import Any3from typing import List4from typing import Optional5from typing import Tuple6from typing import Union789def convert_row(10values: Optional[Union[List[Any], Tuple[Any, ...]]],11converters: List[Any],12) -> Optional[Tuple[Any, ...]]:13"""14Convert a row of data values.1516Parameters17----------18values : tuple or None19Tuple containing values in a row of data20converters : list[tuple]21List of two-element tuples containing a column index and a converter22function. The column index specifies which column to apply the function to.2324Returns25-------26tuple or None2728"""29if values is None:30return None31if not converters:32return tuple(values)33values = list(values)34for i, conv in enumerate(converters):35idx, encoding, func = conv36value = values[idx]37if encoding is not None and value is not None:38value = value.decode(encoding)39if func is not None:40values[idx] = func(value)41else:42values[idx] = value43return tuple(values)444546def convert_rows(rows: List[Any], converters: List[Any]) -> List[Any]:47"""48Convert rows of data values.4950Parameters51----------52rows : list of tuples or None53Rows of data from a query54converters : list[tuple]55List of two-element tuples containing a column index and a converter56function. The column index specifies which column to apply the function to.5758Returns59-------60list of tuples6162"""63if not rows or not converters:64return rows65rows = list(rows)66for i, row in enumerate(rows):67rows[i] = convert_row(row, converters=converters)68return rows697071