#-----------------------------------------------------------------------------
# Copyright (c) Anaconda, Inc., and Bokeh Contributors.
# All rights reserved.
#
# The full license is in the file LICENSE.txt, distributed with this software.
#-----------------------------------------------------------------------------
"""
"""
#-----------------------------------------------------------------------------
# Boilerplate
#-----------------------------------------------------------------------------
from __future__ import annotations
import logging # isort:skip
log = logging.getLogger(__name__)
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
# Standard library imports
from collections.abc import (
Container,
Iterable,
Mapping,
Sequence,
Sized,
)
from typing import TYPE_CHECKING, Any, TypeVar
# Bokeh imports
from ._sphinx import property_link, register_type_link, type_link
from .bases import (
ContainerProperty,
Init,
Property,
SingleParameterizedProperty,
TypeOrInst,
)
from .descriptors import ColumnDataPropertyDescriptor
from .enum import Enum
from .numeric import Int
from .singletons import Intrinsic, Undefined
from .wrappers import (
PropertyValueColumnData,
PropertyValueDict,
PropertyValueList,
PropertyValueSet,
)
if TYPE_CHECKING:
from ...document.events import DocumentPatchedEvent
#-----------------------------------------------------------------------------
# Globals and constants
#-----------------------------------------------------------------------------
__all__ = (
'Array',
'ColumnData',
'Dict',
'Len',
'List',
'NonEmpty',
'RelativeDelta',
'RestrictedDict',
'Seq',
'Set',
'Tuple',
)
T = TypeVar("T")
#-----------------------------------------------------------------------------
# General API
#-----------------------------------------------------------------------------
[docs]
class Seq(ContainerProperty[T]):
""" Accept non-string ordered sequences of values, e.g. list, tuple, array.
"""
def __init__(self, item_type: TypeOrInst[Property[T]], *, default: Init[T] = Undefined, help: str | None = None) -> None:
super().__init__(item_type, default=default, help=help)
@property
def item_type(self):
return self.type_params[0]
def validate(self, value: Any, detail: bool = True) -> None:
super().validate(value, True)
if self._is_seq(value) and all(self.item_type.is_valid(item) for item in value):
return
if self._is_seq(value):
invalid = []
for item in value:
if not self.item_type.is_valid(item):
invalid.append(item)
msg = "" if not detail else f"expected an element of {self}, got seq with invalid items {invalid!r}"
raise ValueError(msg)
msg = "" if not detail else f"expected an element of {self}, got {value!r}"
raise ValueError(msg)
@classmethod
def _is_seq(cls, value: Any) -> bool:
return ((isinstance(value, Sequence) or cls._is_seq_like(value)) and not isinstance(value, str))
@classmethod
def _is_seq_like(cls, value: Any) -> bool:
return (isinstance(value, Container | Sized | Iterable)
and hasattr(value, "__getitem__") # NOTE: this is what makes it disallow set type
and not isinstance(value, Mapping))
[docs]
class List(Seq[T]):
""" Accept Python list values.
"""
def __init__(self, item_type: TypeOrInst[Property[T]], *, default: Init[T] = [], help: str | None = None) -> None:
# TODO: refactor to not use mutable objects as default values.
# Left in place for now because we want to allow None to express
# optional values. Also in Dict.
super().__init__(item_type, default=default, help=help)
def wrap(self, value: list[T]) -> PropertyValueList[T]:
""" Some property types need to wrap their values in special containers, etc.
"""
if isinstance(value, list):
if isinstance(value, PropertyValueList):
return value
else:
return PropertyValueList(value)
else:
return value
@classmethod
def _is_seq(cls, value: Any):
return isinstance(value, list)
[docs]
class Set(Seq[T]):
""" Accept Python ``set()`` values.
"""
def __init__(self, item_type: TypeOrInst[Property[T]], *, default: Init[T] = set(), help: str | None = None) -> None:
# TODO: refactor to not use mutable objects as default values.
# Left in place for now because we want to allow None to express
# optional values. Also in Dict.
super().__init__(item_type, default=default, help=help)
def wrap(self, value: set[T]) -> PropertyValueSet[T]:
""" Some property types need to wrap their values in special containers, etc. """
if isinstance(value, set):
if isinstance(value, PropertyValueSet):
return value
else:
return PropertyValueSet(value)
else:
return value
@classmethod
def _is_seq(cls, value: Any) -> bool:
return isinstance(value, set)
[docs]
class Array(Seq[T]):
""" Accept NumPy array values.
"""
@classmethod
def _is_seq(cls, value: Any) -> bool:
import numpy as np
return isinstance(value, np.ndarray)
[docs]
class Dict(ContainerProperty[Any]):
""" Accept Python dict values.
If a default value is passed in, then a shallow copy of it will be
used for each new use of this property.
"""
def __init__(self, keys_type: TypeOrInst[Property[Any]], values_type: TypeOrInst[Property[Any]], *,
default: Init[T] = {}, help: str | None = None) -> None:
super().__init__(keys_type, values_type, default=default, help=help)
@property
def keys_type(self):
return self.type_params[0]
@property
def values_type(self):
return self.type_params[1]
def validate(self, value: Any, detail: bool = True) -> None:
super().validate(value, detail)
key_is_valid = self.keys_type.is_valid
value_is_valid = self.values_type.is_valid
expected = f"expected a dict of type {self}"
if not isinstance(value, dict):
raise ValueError(f"{expected}, got a value of type {type(value)}" if detail else "")
bad_keys = [str(k) for k in value if not key_is_valid(k)]
bad_value_keys = [str(k) for (k, v) in value.items() if not value_is_valid(v)]
exception_header = f"{expected}, got a dict with"
bad_keys_str = f"invalid keys: {', '.join(bad_keys)}"
bad_value_keys_str = f"invalid values for keys: {', '.join(bad_value_keys)}"
err = None
if (has_bad_keys := any(bad_keys)) & (has_bad_key_values := any(bad_value_keys)):
err = ValueError(f"{exception_header} {bad_keys_str} and {bad_value_keys_str}")
elif has_bad_keys:
err = ValueError(f"{exception_header} {bad_keys_str}")
elif has_bad_key_values:
err = ValueError(f"{exception_header} {bad_value_keys_str}")
if err:
raise err if detail else ValueError("")
def wrap(self, value):
""" Some property types need to wrap their values in special containers, etc.
"""
if isinstance(value, dict):
if isinstance(value, PropertyValueDict):
return value
else:
return PropertyValueDict(value)
else:
return value
[docs]
class ColumnData(Dict):
""" Accept a Python dictionary suitable as the ``data`` attribute of a
:class:`~bokeh.models.sources.ColumnDataSource`.
This class is a specialization of ``Dict`` that handles efficiently
encoding columns that are NumPy arrays.
"""
def make_descriptors(self, base_name):
""" Return a list of ``ColumnDataPropertyDescriptor`` instances to
install on a class, in order to delegate attribute access to this
property.
Args:
base_name (str) : the name of the property these descriptors are for
Returns:
list[ColumnDataPropertyDescriptor]
The descriptors returned are collected by the ``MetaHasProps``
metaclass and added to ``HasProps`` subclasses during class creation.
"""
return [ ColumnDataPropertyDescriptor(base_name, self) ]
def _hinted_value(self, value: Any, hint: DocumentPatchedEvent | None) -> Any:
from ...document.events import ColumnDataChangedEvent, ColumnsStreamedEvent
if isinstance(hint, ColumnDataChangedEvent):
return { col: hint.model.data[col] for col in hint.cols }
if isinstance(hint, ColumnsStreamedEvent):
return hint.data
return value
def wrap(self, value):
""" Some property types need to wrap their values in special containers, etc.
"""
if isinstance(value, dict):
if isinstance(value, PropertyValueColumnData):
return value
else:
return PropertyValueColumnData(value)
else:
return value
[docs]
class Tuple(ContainerProperty):
""" Accept Python tuple values.
"""
def __init__(self, *type_params: TypeOrInst[Property[Any]], default: Init[T] = Undefined, help: str | None = None) -> None:
super().__init__(*type_params, default=default, help=help)
def validate(self, value: Any, detail: bool = True) -> None:
super().validate(value, detail)
if isinstance(value, tuple | list) and len(self.type_params) == len(value):
if all(type_param.is_valid(item) for type_param, item in zip(self.type_params, value)):
return
msg = "" if not detail else f"expected an element of {self}, got {value!r}"
raise ValueError(msg)
def transform(self, value):
""" Change the value into a JSON serializable format.
"""
return tuple(typ.transform(x) for (typ, x) in zip(self.type_params, value))
[docs]
class RelativeDelta(Dict):
""" Accept RelativeDelta dicts for time delta values.
"""
def __init__(self, default={}, *, help: str | None = None) -> None:
keys = Enum("years", "months", "days", "hours", "minutes", "seconds", "microseconds")
values = Int
super().__init__(keys, values, default=default, help=help)
def __str__(self) -> str:
return self.__class__.__name__
[docs]
class RestrictedDict(Dict):
""" Check for disallowed key(s).
"""
def __init__(self, keys_type, values_type, disallow, default={}, *, help: str | None = None) -> None:
self._disallow = set(disallow)
super().__init__(keys_type=keys_type, values_type=values_type, default=default, help=help)
def validate(self, value: Any, detail: bool = True) -> None:
super().validate(value, detail)
error_keys = self._disallow & value.keys()
if error_keys:
msg = "" if not detail else f"Disallowed keys: {error_keys!r}"
raise ValueError(msg)
TSeq = TypeVar("TSeq", bound=Seq[Any])
class NonEmpty(SingleParameterizedProperty[TSeq]):
""" Allows only non-empty containers. """
def __init__(self, type_param: TypeOrInst[TSeq], *, default: Init[TSeq] = Intrinsic,
help: str | None = None) -> None:
super().__init__(type_param, default=default, help=help)
def validate(self, value: Any, detail: bool = True) -> None:
super().validate(value, detail)
if not value:
msg = "" if not detail else "Expected a non-empty container"
raise ValueError(msg)
class Len(SingleParameterizedProperty[TSeq]):
""" Allows only containers of the given length. """
def __init__(self, type_param: TypeOrInst[TSeq], length: int, *, default: Init[TSeq] = Intrinsic,
help: str | None = None) -> None:
super().__init__(type_param, default=default, help=help)
self.length = length
def validate(self, value: Any, detail: bool = True) -> None:
super().validate(value, detail)
if len(value) != self.length:
msg = "" if not detail else f"Expected a container of length #{self.length}, got #{len(value)}"
raise ValueError(msg)
#-----------------------------------------------------------------------------
# Dev API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Private API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------
@register_type_link(Dict)
def _sphinx_type_dict(obj: Dict):
return f"{property_link(obj)}({type_link(obj.keys_type)}, {type_link(obj.values_type)})"
@register_type_link(Seq)
def _sphinx_type_seq(obj: Seq[Any]):
return f"{property_link(obj)}({type_link(obj.item_type)})"
@register_type_link(Tuple)
def _sphinx_type_tuple(obj: Tuple):
item_types = ", ".join(type_link(x) for x in obj.type_params)
return f"{property_link(obj)}({item_types})"
