Skip to content

Commit

Permalink
More types
Browse files Browse the repository at this point in the history
  • Loading branch information
philippjfr committed Nov 15, 2024
1 parent e051df1 commit 95e0b52
Show file tree
Hide file tree
Showing 3 changed files with 52 additions and 25 deletions.
53 changes: 39 additions & 14 deletions panel/_param.py
Original file line number Diff line number Diff line change
@@ -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):
"""
Expand All @@ -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', "
Expand All @@ -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(
Expand All @@ -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)):
Expand All @@ -69,27 +89,32 @@ 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 '
f'tuple values of length 2 (vertical, horizontal) or 4 '
'(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
6 changes: 3 additions & 3 deletions panel/io/cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
18 changes: 10 additions & 8 deletions panel/io/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,16 @@
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
from ..util import escape
from .state import state

if TYPE_CHECKING:
from pyinstrument.session import Session

_P = ParamSpec("_P")
_R = TypeVar("_R")

Expand Down Expand Up @@ -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
Expand All @@ -219,22 +221,22 @@ 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}'
tracker = memray.Tracker(tmp_file)
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())
Expand Down Expand Up @@ -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
Expand Down

0 comments on commit 95e0b52

Please sign in to comment.