Spaces:
Running
Running
File size: 1,751 Bytes
1380717 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
from __future__ import annotations
from typing import TYPE_CHECKING, Callable, overload
from altair.utils.core import sanitize_pandas_dataframe
from altair.utils.data import DataTransformerRegistry as _DataTransformerRegistry
from altair.utils.data import (
MaxRowsError,
check_data_type,
limit_rows,
sample,
to_csv,
to_json,
to_values,
)
if TYPE_CHECKING:
from altair.utils.data import DataType, ToValuesReturnType
from altair.utils.plugin_registry import PluginEnabler
@overload
def default_data_transformer(
data: None = ..., max_rows: int = ...
) -> Callable[[DataType], ToValuesReturnType]: ...
@overload
def default_data_transformer(
data: DataType, max_rows: int = ...
) -> ToValuesReturnType: ...
def default_data_transformer(
data: DataType | None = None, max_rows: int = 5000
) -> Callable[[DataType], ToValuesReturnType] | ToValuesReturnType:
if data is None:
def pipe(data: DataType, /) -> ToValuesReturnType:
data = limit_rows(data, max_rows=max_rows)
return to_values(data)
return pipe
else:
return to_values(limit_rows(data, max_rows=max_rows))
class DataTransformerRegistry(_DataTransformerRegistry):
def disable_max_rows(self) -> PluginEnabler:
"""Disable the MaxRowsError."""
options = self.options
if self.active in {"default", "vegafusion"}:
options = options.copy()
options["max_rows"] = None
return self.enable(**options)
__all__ = (
"DataTransformerRegistry",
"MaxRowsError",
"check_data_type",
"default_data_transformer",
"limit_rows",
"sample",
"sanitize_pandas_dataframe",
"to_csv",
"to_json",
"to_values",
)
|