diff --git a/panel/_param.py b/panel/_param.py index 0c8e0c72f8..3a579f8569 100644 --- a/panel/_param.py +++ b/panel/_param.py @@ -1,5 +1,21 @@ +from __future__ import annotations + +from enum import Enum +from typing import ( + TYPE_CHECKING, Any, Literal, TypeAlias, cast, +) + from param import Parameter, _is_number +if TYPE_CHECKING: + MarginType: TypeAlias = int | tuple[int, int] | tuple[int, int, int] | tuple[int, int, int, int] + +class AlignmentEnum(Enum): + AUTO = 'auto' + START = 'start' + CENTER = 'center' + END = 'end' + class Align(Parameter): """ @@ -8,16 +24,20 @@ class Align(Parameter): to the (vertical, horizontal) alignment. """ - def __init__(self, default='start', **params): + def __init__( + self, + default: AlignmentEnum | tuple[AlignmentEnum, AlignmentEnum] = AlignmentEnum.START, + **params: Any + ): super().__init__(default=default, **params) self._validate(default) - def _validate(self, val): + def _validate(self, val: Any) -> None: self._validate_value(val, self.allow_None) - def _validate_value(self, val, allow_None, valid=('auto', 'start', 'center', 'end')): - if ((val is None and allow_None) or val in valid or - (isinstance(val, tuple) and len(val) == 2 and all(v in valid for v in val))): + def _validate_value(self, val: Any, allow_None: bool) -> None: + if ((val is None and allow_None) or val in AlignmentEnum or + (isinstance(val, tuple) and len(val) == 2 and all(v in AlignmentEnum for v in val))): return raise ValueError( f"Align parameter {self.name!r} must be one of 'start', " @@ -36,10 +56,10 @@ def __init__(self, default=None, allow_None=True, **params): super().__init__(default=default, allow_None=allow_None, **params) self._validate(default) - def _validate(self, val): + def _validate(self, val: Any) -> None: self._validate_value(val, self.allow_None) - def _validate_value(self, val, allow_None): + def _validate_value(self, val: Any, allow_None: bool) -> None: if (val is None and allow_None) or val == 'auto' or _is_number(val): return raise ValueError( @@ -60,7 +80,7 @@ def __init__(self, default=None, allow_None=True, **params): super().__init__(default=default, allow_None=allow_None, **params) self._validate(default) - def _validate_value(self, val, allow_None): + def _validate_value(self, val: Any, allow_None: bool) -> None: if val is None and allow_None: return if not isinstance(val, (tuple, int)): @@ -69,8 +89,8 @@ def _validate_value(self, val, allow_None): f'tuple values, not values of not {type(val)!r}.' ) - def _validate_length(self, val): - if not isinstance(val, tuple) or len(val) in (2, 4): + def _validate_length(self, val: Any) -> None: + if not isinstance(val, tuple) or (1 < len(val) < 5): return raise ValueError( f'Margin parameter {self.name!r} only takes integer and ' @@ -78,18 +98,23 @@ def _validate_length(self, val): '(top, right, bottom, left).' ) - def _validate(self, val): + def _validate(self, val: Any) -> None: self._validate_value(val, self.allow_None) self._validate_length(val) @classmethod - def serialize(cls, value): + def serialize(cls, value: MarginType) -> Literal['null'] | list[int] | int: if value is None: return 'null' return list(value) if isinstance(value, tuple) else value @classmethod - def deserialize(cls, value): + def deserialize(cls, value: Literal['null'] | list[int] | int) -> MarginType | None: if value == 'null': return None - return tuple(value) if isinstance(value, list) else value + elif isinstance(value, list): + n = len(value) + if (n < 2 or n > 5): + raise ValueError('Cannot deserialize list of length {n).') + return cast(MarginType, tuple(value)) + return value diff --git a/panel/io/cache.py b/panel/io/cache.py index 59a1d83bc4..6003840883 100644 --- a/panel/io/cache.py +++ b/panel/io/cache.py @@ -18,8 +18,8 @@ from contextlib import contextmanager from typing import ( - TYPE_CHECKING, Any, Callable, Hashable, Literal, ParamSpec, Protocol, - TypeVar, overload, + TYPE_CHECKING, Any, Awaitable, Callable, Hashable, Literal, ParamSpec, + Protocol, TypeVar, cast, overload, ) import param @@ -519,7 +519,7 @@ async def wrapped_func(*args: _P.args, **kwargs: _P.kwargs) -> _R: ret, ts, count, _ = func_cache[hash_value] func_cache[hash_value] = (ret, ts, count+1, time) else: - ret = await func(*args, **kwargs) + ret = await cast(Awaitable[Any], func(*args, **kwargs)) with lock: func_cache[hash_value] = (ret, time, 0, time) return ret diff --git a/panel/io/profile.py b/panel/io/profile.py index fb39f1d0c6..37495bbb22 100644 --- a/panel/io/profile.py +++ b/panel/io/profile.py @@ -10,7 +10,7 @@ from cProfile import Profile from functools import wraps from typing import ( - TYPE_CHECKING, Callable, Iterator, Literal, ParamSpec, TypeVar, + TYPE_CHECKING, Callable, Iterator, Literal, ParamSpec, Sequence, TypeVar, ) from ..config import config @@ -18,6 +18,8 @@ from .state import state if TYPE_CHECKING: + from pyinstrument.session import Session + _P = ParamSpec("_P") _R = TypeVar("_R") @@ -194,7 +196,7 @@ def update_memray(*args): @contextmanager -def profile_ctx(engine: ProfilingEngine = 'pyinstrument') -> Iterator[list[Profile | bytes]]: +def profile_ctx(engine: ProfilingEngine = 'pyinstrument') -> Iterator[Sequence[Profile | bytes | Session]]: """ A context manager which profiles the body of the with statement with the supplied profiling engine and returns the profiling object @@ -219,8 +221,8 @@ def profile_ctx(engine: ProfilingEngine = 'pyinstrument') -> Iterator[list[Profi prof = Profiler(async_mode='disabled') prof.start() elif engine == 'snakeviz': - prof = Profile() - prof.enable() + profile = Profile() + profile.enable() elif engine == 'memray': import memray tmp_file = f'{tempfile.gettempdir()}/tmp{uuid.uuid4().hex}' @@ -228,13 +230,13 @@ def profile_ctx(engine: ProfilingEngine = 'pyinstrument') -> Iterator[list[Profi tracker.__enter__() elif engine is None: pass - sessions: list[Profile | bytes] = [] + sessions: Sequence[Profile | bytes | Session] = [] yield sessions if engine == 'pyinstrument': sessions.append(prof.stop()) elif engine == 'snakeviz': - prof.disable() - sessions.append(prof) + profile.disable() + sessions.append(profile) elif engine == 'memray': tracker.__exit__(None, None, None) sessions.append(open(tmp_file, 'rb').read()) @@ -262,7 +264,7 @@ def wrapped(*args: _P.args, **kwargs: _P.kwargs) -> _R: return func(*args, **kwargs) with profile_ctx(engine) as sessions: ret = func(*args, **kwargs) - state._profiles[(name, engine)] += sessions + state._profiles[(name, engine)] += list(sessions) state.param.trigger('_profiles') return ret return wrapped