More types

panel/chat/feed.py

@@ -410,7 +410,7 @@ def _upsert_message(
         is_stopped = self._callback_future is not None and self._callback_future.cancelled()
         if value is None:
             # don't add new message if the callback returns None
-            return
+            return None
         elif is_stopping or is_stopped:
             raise StopCallback("Callback was stopped.")
 
@@ -492,12 +492,14 @@ async def _serialize_response(self, response: Any) -> ChatMessage | None:
                 self._callback_state = CallbackState.GENERATING
                 async for token in response:
                     response_message = self._upsert_message(token, response_message)
-                    response_message.show_activity_dot = self.show_activity_dot
+                    if response_message is not None:
+                        response_message.show_activity_dot = self.show_activity_dot
             elif isgenerator(response):
                 self._callback_state = CallbackState.GENERATING
                 for token in response:
                     response_message = self._upsert_message(token, response_message)
-                    response_message.show_activity_dot = self.show_activity_dot
+                    if response_message is not None:
+                        response_message.show_activity_dot = self.show_activity_dot
             elif isawaitable(response):
                 response_message = self._upsert_message(await response, response_message)
             else:
@@ -528,7 +530,7 @@ async def _schedule_placeholder(
                 return
             await asyncio.sleep(0.1)
 
-    async def _handle_callback(self, message, loop: asyncio.BaseEventLoop):
+    async def _handle_callback(self, message, loop: asyncio.AbstractEventLoop):
         callback_args, callback_kwargs = self._gather_callback_args(message)
         if iscoroutinefunction(self.callback):
             response = await self.callback(*callback_args, **callback_kwargs)
@@ -562,16 +564,16 @@ async def _prepare_response(self, *_) -> None:
 
             num_entries = len(self._chat_log)
             loop = asyncio.get_event_loop()
-            future = loop.create_task(self._handle_callback(message, loop))
-            self._callback_future = future
+            task = loop.create_task(self._handle_callback(message, loop))
+            self._callback_future = task
             await asyncio.gather(
-                self._schedule_placeholder(future, num_entries), future,
+                self._schedule_placeholder(task, num_entries), task,
             )
         except StopCallback:
             # callback was stopped by user
             self._callback_state = CallbackState.STOPPED
         except Exception as e:
-            send_kwargs = dict(user="Exception", respond=False)
+            send_kwargs: dict[str, Any] = dict(user="Exception", respond=False)
             if self.callback_exception == "summary":
                 self.send(
                     f"Encountered `{e!r}`. "
@@ -634,7 +636,7 @@ def send(
                     "Cannot set user or avatar when explicitly sending "
                     "a ChatMessage. Set them directly on the ChatMessage."
                 )
-            message = value
+            message: ChatMessage | None = value
         else:
             if not isinstance(value, dict):
                 value = {"object": value}
@@ -866,7 +868,7 @@ async def _prepare_prompt(*_) -> None:
                 send_kwargs["user"] = "Input"
             self.send(form, respond=False, **send_kwargs)
 
-            for _ in range(timeout * 10):  # sleeping for 0.1 seconds
+            for _1 in range(timeout * 10):  # sleeping for 0.1 seconds
                 is_fulfilled = predicate(component) if predicate else True
                 submit_button.disabled = not is_fulfilled
                 if submit_button.clicks > 0:

panel/chat/interface.py

@@ -16,6 +16,7 @@
 
 from ..io.resources import CDN_DIST
 from ..layout import Row, Tabs
+from ..layout.base import ListLike, NamedListLike
 from ..pane.image import ImageBase
 from ..viewable import Viewable
 from ..widgets.base import WidgetBase
@@ -189,7 +190,7 @@ def _link_disabled_loading(self, obj: Viewable):
         """
         for attr in ["disabled", "loading"]:
             setattr(obj, attr, getattr(self, attr))
-            self.link(obj, **{attr: attr})
+            self.link(obj, callbacks=None, bidirectional=False, **{attr: attr})
 
     @param.depends("width", watch=True)
     def _update_input_width(self):
@@ -549,7 +550,11 @@ def active_widget(self) -> WidgetBase:
         The active widget.
         """
         if isinstance(self._input_layout, Tabs):
-            return self._input_layout[self.active].objects[0]
+            current_tab = self._input_layout[self.active]
+            if isinstance(current_tab, (ListLike, NamedListLike)):
+                return current_tab.objects[0]
+            else:
+                return current_tab  # type: ignore
         return self._input_layout.objects[0]
 
     @property
@@ -585,7 +590,7 @@ def _serialize_for_transformers(
         messages: list[ChatMessage],
         role_names: dict[str, str | list[str]] | None = None,
         default_role: str | None = "assistant",
-        custom_serializer: Callable = None,
+        custom_serializer: Callable[[ChatMessage], Any] | None = None,
         **serialize_kwargs
     ) -> list[dict[str, Any]]:
         """

panel/chat/message.py

@@ -13,7 +13,7 @@
 from io import BytesIO
 from tempfile import NamedTemporaryFile
 from typing import (
-    TYPE_CHECKING, Any, ClassVar, Union,
+    TYPE_CHECKING, Any, ClassVar, TypedDict, Union,
 )
 from zoneinfo import ZoneInfo
 
@@ -31,20 +31,26 @@
 )
 from ..pane.media import Audio, Video
 from ..param import ParamFunction
-from ..viewable import Viewable
+from ..viewable import ServableMixin, Viewable
 from ..widgets.base import Widget
 from .icon import ChatCopyIcon, ChatReactionIcons
 from .utils import (
     avatar_lookup, build_avatar_pane, serialize_recursively, stream_to,
 )
 
+Avatar = Union[str, BytesIO, bytes, ImageBase]
+AvatarDict = dict[str, Avatar]
+
 if TYPE_CHECKING:
     from bokeh.document import Document
     from bokeh.model import Model
     from pyviz_comms import Comm
 
-Avatar = Union[str, BytesIO, bytes, ImageBase]
-AvatarDict = dict[str, Avatar]
+    class MessageParams(TypedDict, total=False):
+        avatar: Avatar
+        user: str
+        object: Any
+        value: Any
 
 USER_LOGO = "🧑"
 ASSISTANT_LOGO = "🤖"
@@ -385,11 +391,11 @@ def _select_renderer(
         self,
         contents: Any,
         mime_type: str,
-    ):
+    ) -> tuple[Any, type[Pane] | Callable[..., Pane | ServableMixin]]:
         """
         Determine the renderer to use based on the mime type.
         """
-        renderer = _panel
+        renderer: type[Pane] | Callable[..., Pane | ServableMixin] = _panel
         if mime_type == "application/pdf":
             contents = self._exit_stack.enter_context(BytesIO(contents))
             renderer = partial(PDF, embed=True)
@@ -628,7 +634,7 @@ def stream(self, token: str, replace: bool = False):
 
     def update(
         self,
-        value: dict | ChatMessage | Any,
+        value: MessageParams | ChatMessage | Any,
         user: str | None = None,
         avatar: str | bytes | BytesIO | None = None,
     ):
@@ -644,9 +650,9 @@ def update(
         avatar : str | bytes | BytesIO | None
             The avatar to use; overrides the message message's avatar if provided.
         """
-        updates = {}
+        updates: MessageParams = {}
         if isinstance(value, dict):
-            updates.update(value)
+            updates.update(value)  # type: ignore
             if user:
                 updates["user"] = user
             if avatar:

panel/command/__init__.py

@@ -77,20 +77,20 @@ def main(args: list[str] | None = None):
     for cls in bokeh_commands:
         if cls is BkServe:
             subparser = subs.add_parser(Serve.name, help=Serve.help)
-            subcommand = Serve(parser=subparser)
-            subparser.set_defaults(invoke=subcommand.invoke)
+            serve_subcommand = Serve(parser=subparser)
+            subparser.set_defaults(invoke=serve_subcommand.invoke)
         elif cls is Compile:
             subparser = subs.add_parser(Compile.name, help=Compile.help)
-            subcommand = Compile(parser=subparser)
-            subparser.set_defaults(invoke=subcommand.invoke)
+            compile_subcommand = Compile(parser=subparser)
+            subparser.set_defaults(invoke=compile_subcommand.invoke)
         elif cls is Convert:
             subparser = subs.add_parser(Convert.name, help=Convert.help)
-            subcommand = Convert(parser=subparser)
-            subparser.set_defaults(invoke=subcommand.invoke)
+            convert_subcommand = Convert(parser=subparser)
+            subparser.set_defaults(invoke=convert_subcommand.invoke)
         elif cls is Bundle:
             subparser = subs.add_parser(Bundle.name, help=Bundle.help)
-            subcommand = Bundle(parser=subparser)
-            subparser.set_defaults(invoke=subcommand.invoke)
+            bundle_subcommand = Bundle(parser=subparser)
+            subparser.set_defaults(invoke=bundle_subcommand.invoke)
         else:
             subs.add_parser(cls.name, help=cls.help)
 
@@ -99,8 +99,8 @@ def main(args: list[str] | None = None):
         die(f"ERROR: Must specify subcommand, one of: {nice_join(all_commands)}")
 
     if sys.argv[1] in ('--help', '-h'):
-        args = parser.parse_args(sys.argv[1:])
-        args.invoke(args)
+        parser = parser.parse_args(sys.argv[1:])
+        parser.invoke(args)
         sys.exit()
 
     if len(sys.argv) > 1 and any(sys.argv[1] == c.name for c in bokeh_commands):

panel/io/convert.py

@@ -169,7 +169,7 @@ def script_to_html(
     ---------
     filename: str | Path | IO
         The filename of the Panel/Bokeh application to convert.
-    requirements: 'auto' | List[str]
+    requirements: 'auto' | List[str] | os.PathLike
         The list of requirements to include (in addition to Panel).
     js_resources: 'auto' | List[str]
         The list of JS resources to include in the exported HTML.
@@ -215,19 +215,23 @@ def script_to_html(
         )
 
     if requirements == 'auto':
-        requirements = find_requirements(source)
+        requirement_list = find_requirements(source)
     elif isinstance(requirements, str) and pathlib.Path(requirements).is_file():
-        requirements = pathlib.Path(requirements).read_text(encoding='utf-8').splitlines()
+        requirement_list = pathlib.Path(requirements).read_text(encoding='utf-8').splitlines()
         try:
             from packaging.requirements import Requirement
-            requirements = [
-                r2 for r in requirements
+            requirement_list = [
+                r2 for r in requirement_list
                 if (r2 := r.split("#")[0].strip()) and Requirement(r2)
             ]
         except Exception as e:
             raise ValueError(
                 f'Requirements parser raised following error: {e}'
             ) from e
+    elif isinstance(requirements, list):
+        requirement_list = requirements
+    else:
+        raise ValueError(f'Could not resolve requirements file {requirements}')
 
     # Environment
     if panel_version == 'local':
@@ -244,7 +248,7 @@ def script_to_html(
     if http_patch:
         base_reqs.append('pyodide-http==0.2.1')
     reqs = base_reqs + [
-        req for req in requirements if req not in ('panel', 'bokeh')
+        req for req in requirement_list if req not in ('panel', 'bokeh')
     ]
     for name, min_version in MINIMUM_VERSIONS.items():
         if any(name in req for req in reqs):
@@ -554,7 +558,10 @@ def convert_apps(
     }
 
     if state._is_pyodide:
-        files = {app: convert_app(app, dest_path, requirements=app_requirements, **kwargs) for app in apps}
+        files = {
+            app: convert_app(app, dest_path, requirements=app_requirements, **kwargs)  # type: ignore
+            for app in apps
+        }
     else:
         files = _convert_process_pool(
             apps, dest_path, max_workers=max_workers, requirements=app_requirements, **kwargs

panel/io/pyodide.py

@@ -44,6 +44,9 @@
 if TYPE_CHECKING:
     from bokeh.core.types import ID
 
+    from ..template.base import TemplateBase
+    from ..viewable import Viewable
+
 try:
     from js import document as js_document  # noqa
     try:
@@ -166,28 +169,30 @@ def _doc_json(doc: Document, root_els=None) -> tuple[str, str, str]:
         })
     return json.dumps(docs_json), json.dumps(render_items_json), json.dumps(root_ids)
 
-def _model_json(model: Model, target: str) -> tuple[Document, str]:
+def _model_json(viewable: Viewable | TemplateBase, target: str) -> tuple[Document, str]:
     """
     Renders a Bokeh Model to JSON representation given a particular
     DOM target and returns the Document and the serialized JSON string.
 
     Arguments
     ---------
-    model: bokeh.model.Model
+    model: Viewable
         The bokeh model to render.
     target: str
         The id of the DOM node to render to.
 
     Returns
     -------
+    viewable: The Viewable to render to JSON
+        The viewable to render
     document: Document
         The bokeh Document containing the rendered Bokeh Model.
     model_json: str
         The serialized JSON representation of the Bokeh Model.
     """
     doc = Document()
     doc.hold()
-    model.server_doc(doc=doc)
+    viewable.server_doc(doc=doc)
     model = doc.roots[0]
     docs_json, _ = standalone_docs_json_and_render_items(
         [model], suppress_callback_warning=True

panel/io/resources.py

@@ -44,13 +44,20 @@
 if TYPE_CHECKING:
     from bokeh.resources import Urls
 
-    # Make fields optional
+    class TarballType(TypedDict, total=False):
+        tar: str
+        src: str
+        dest: str
+        exclude: list[str]
+
     class ResourcesType(TypedDict, total=False):
         css: dict[str, str]
         font: dict[str, str]
         js: dict[str, str]
         js_modules: dict[str, str]
         raw_css: list[str]
+        tarball: dict[str, TarballType]
+        bundle: bool
 
 logger = logging.getLogger(__name__)
 

panel/io/server.py

@@ -612,7 +612,7 @@ class ComponentResourceHandler(StaticFileHandler):
     ]
 
     def initialize(self, path: str | None = None, default_filename: str | None = None):
-        self.root = path
+        self.root = path or 'root'
         self.default_filename = default_filename
 
     def parse_url_path(self, path: str) -> str:

panel/pane/base.py

@@ -143,7 +143,7 @@ class PaneBase(Layoutable):
     # Whether applies requires full set of keywords
     _applies_kw: ClassVar[bool] = False
 
-    _skip_layoutable = ('css_classes', 'margin', 'name')
+    _skip_layoutable: ClassVar[tuple[str, ...]] = ('css_classes', 'margin', 'name')
 
     # Whether the Pane layout can be safely unpacked
     _unpack: ClassVar[bool] = True
@@ -530,14 +530,18 @@ class ModelPane(Pane):
     `bokeh.model.Model` can consume.
     """
 
-    _bokeh_model: ClassVar[type[Model]] | None = None
+    _bokeh_model: ClassVar[type[Model] | None] = None
 
     __abstract = True
 
     def _get_model(
         self, doc: Document, root: Model | None = None,
         parent: Model | None = None, comm: Comm | None = None
     ) -> Model:
+        if self._bokeh_model is None:
+            raise NotImplementedError(
+                'Pane {type(self).__name__} did not define a _bokeh_model'
+            )
         model = self._bokeh_model(**self._get_properties(doc))
         if root is None:
             root = model