Add types to panel.io (#3505)

panel/io/document.py

@@ -7,17 +7,16 @@
 
 from contextlib import contextmanager
 from functools import partial, wraps
-from typing import Callable, List, Optional
+from typing import Callable, Iterator, List, Optional
 
 from bokeh.application.application import SessionContext
 from bokeh.document.document import Document
 from bokeh.document.events import DocumentChangedEvent, ModelChangedEvent
 from bokeh.io import curdoc as _curdoc
 
-from .model import patch_events
+from .model import monkeypatch_events
 from .state import set_curdoc, state
 
-
 #---------------------------------------------------------------------
 # Private API
 #---------------------------------------------------------------------
@@ -61,6 +60,8 @@ def _dispatch_events(doc: Document, events: List[DocumentChangedEvent]) -> None:
 
 def init_doc(doc: Optional[Document]) -> Document:
     curdoc = doc or _curdoc()
+    if not isinstance(curdoc, Document):
+        curdoc = curdoc._doc
     if not curdoc.session_context:
         return curdoc
 
@@ -107,18 +108,20 @@ def wrapper(*args, **kw):
     return wrapper
 
 @contextmanager
-def unlocked():
+def unlocked() -> Iterator:
     """
     Context manager which unlocks a Document and dispatches
     ModelChangedEvents triggered in the context body to all sockets
     on current sessions.
     """
     curdoc = state.curdoc
-    if curdoc is None or curdoc.session_context is None or curdoc.session_context.session is None:
+    session_context = getattr(curdoc, 'session_context', None)
+    session = getattr(session_context, 'session', None)
+    if (curdoc is None or session_context is None or session is None):
         yield
         return
     from tornado.websocket import WebSocketHandler
-    connections = curdoc.session_context.session._subscribed_connections
+    connections = session._subscribed_connections
 
     hold = curdoc.callbacks.hold_value
     if hold:
@@ -137,7 +140,7 @@ def unlocked():
                 break
 
         events = curdoc.callbacks._held_events
-        patch_events(events)
+        monkeypatch_events(events)
         remaining_events = []
         for event in events:
             if not isinstance(event, ModelChangedEvent) or event in old_events or locked:
@@ -147,7 +150,7 @@ def unlocked():
                 socket = conn._socket
                 ws_conn = getattr(socket, 'ws_connection', False)
                 if (not hasattr(socket, 'write_message') or
-                    ws_conn is None or (ws_conn and ws_conn.is_closing())):
+                    ws_conn is None or (ws_conn and ws_conn.is_closing())): # type: ignore
                     continue
                 msg = conn.protocol.create('PATCH-DOC', [event])
                 WebSocketHandler.write_message(socket, msg.header_json)

panel/io/location.py

@@ -1,17 +1,28 @@
 """
 Defines the Location  widget which allows changing the href of the window.
 """
+from __future__ import annotations
 
 import json
 import urllib.parse as urlparse
 
+from typing import (
+    TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional
+)
+
 import param
 
 from ..models.location import Location as _BkLocation
 from ..reactive import Syncable
 from ..util import parse_query
+from .document import init_doc
 from .state import state
 
+if TYPE_CHECKING:
+    from bokeh.document import Document
+    from bokeh.model import Model
+    from pyviz_comms import Comm
+
 
 class Location(Syncable):
     """
@@ -55,7 +66,10 @@ def __init__(self, **params):
         self._syncing = False
         self.param.watch(self._update_synced, ['search'])
 
-    def _get_model(self, doc, root=None, parent=None, comm=None):
+    def _get_model(
+        self, doc: 'Document', root: Optional['Model'] = None,
+        parent: Optional['Model'] = None, comm: Optional['Comm'] = None
+    ) -> 'Model':
         model = _BkLocation(**self._process_param_change(self._init_params()))
         root = root or model
         values = self.param.values()
@@ -64,22 +78,28 @@ def _get_model(self, doc, root=None, parent=None, comm=None):
         self._link_props(model, properties, doc, root, comm)
         return model
 
-    def _get_root(self, doc=None, comm=None):
+    def get_root(
+        self, doc: Optional[Document] = None, comm: Optional[Comm] = None, preprocess: bool = True
+    ) -> 'Model':
+        doc = init_doc(doc)
         root = self._get_model(doc, comm=comm)
         ref = root.ref['id']
         state._views[ref] = (self, root, doc, comm)
         self._documents[doc] = root
         return root
 
-    def _cleanup(self, root):
-        if root.document in self._documents:
-            del self._documents[root.document]
-        ref = root.ref['id']
+    def _cleanup(self, root: Optional['Model']) -> None:
+        if root:
+            if root.document in self._documents:
+                del self._documents[root.document]
+            ref = root.ref['id']
+        else:
+            ref = None
         super()._cleanup(root)
         if ref in state._views:
             del state._views[ref]
 
-    def _update_synced(self, event=None):
+    def _update_synced(self, event: param.parameterized.Event = None) -> None:
         if self._syncing:
             return
         query_params = self.query_params
@@ -106,7 +126,9 @@ def _update_synced(self, event=None):
                 if on_error:
                     on_error(mapped)
 
-    def _update_query(self, *events, query=None):
+    def _update_query(
+        self, *events: param.parameterized.Event, query: Optional[Dict[str, Any]] = None
+    ) -> None:
         if self._syncing:
             return
         serialized = query or {}
@@ -129,15 +151,18 @@ def _update_query(self, *events, query=None):
             self._syncing = False
 
     @property
-    def query_params(self):
+    def query_params(self) -> Dict[str, Any]:
         return parse_query(self.search)
 
-    def update_query(self, **kwargs):
+    def update_query(self, **kwargs: Mapping[str, Any]) -> None:
         query = self.query_params
         query.update(kwargs)
         self.search = '?' + urlparse.urlencode(query)
 
-    def sync(self, parameterized, parameters=None, on_error=None):
+    def sync(
+        self, parameterized: param.Parameterized, parameters: Optional[List[str] | Dict[str, str]] = None,
+        on_error: Optional[Callable[[Dict[str, Any]], None]] = None
+    ) -> None:
         """
         Syncs the parameters of a Parameterized object with the query
         parameters in the URL. If no parameters are supplied all
@@ -177,7 +202,7 @@ def sync(self, parameterized, parameters=None, on_error=None):
             query[name] = v
         self._update_query(query=query)
 
-    def unsync(self, parameterized, parameters=None):
+    def unsync(self, parameterized: param.Parameterized, parameters: Optional[List[str]] = None) -> None:
         """
         Unsyncs the parameters of the Parameterized with the query
         params in the URL. If no parameters are supplied all
@@ -187,7 +212,7 @@ def unsync(self, parameterized, parameters=None):
         ---------
         parameterized (param.Parameterized):
           The Parameterized object to unsync query parameters with
-        parameters (list or dict):
+        parameters (list):
           A list of parameters to unsync.
         """
         matches = [s for s in self._synced if s[0] is parameterized]

panel/io/model.py

@@ -1,64 +1,72 @@
 """
 Utilities for manipulating bokeh models.
 """
+from __future__ import annotations
+
 import textwrap
 
 from contextlib import contextmanager
+from typing import (
+    TYPE_CHECKING, Any, List, Iterable, Optional
+)
+from typing_extensions import Literal
 
 import numpy as np
 
 from bokeh.document import Document
-from bokeh.document.events import ModelChangedEvent, ColumnDataChangedEvent
+from bokeh.document.events import (
+    ColumnDataChangedEvent, DocumentPatchedEvent, ModelChangedEvent
+)
 from bokeh.model import DataModel
 from bokeh.models import Box, ColumnDataSource, Model
 from bokeh.protocol import Protocol
 
 from .state import state
 
+if TYPE_CHECKING:
+    from bokeh.core.enums import HoldPolicyType
+    from bokeh.document.events import DocumentChangedEvent
+    from bokeh.protocol.message import Message
+    from pyviz_comms import Comm
 
 #---------------------------------------------------------------------
 # Private API
 #---------------------------------------------------------------------
 
-
 class comparable_array(np.ndarray):
     """
     Array subclass that allows comparisons.
     """
 
-    def __eq__(self, other):
+    def __eq__(self, other: Any) -> bool:
         return super().__eq__(other).all().item()
 
-    def __ne__(self, other):
+    def __ne__(self, other: Any) -> bool:
         return super().__ne__(other).all().item()
 
-
-def patch_events(events):
+def monkeypatch_events(events: List['DocumentChangedEvent']) -> None:
     """
     Patch events applies patches to events that are to be dispatched
     avoiding various issues in Bokeh.
     """
     for e in events:
         # Patch ColumnDataChangedEvents which reference non-existing columns
-        if (hasattr(e, 'hint') and isinstance(e.hint, ColumnDataChangedEvent)
-            and e.hint.cols is not None):
-            e.hint.cols = None
+        if isinstance(getattr(e, 'hint', None), ColumnDataChangedEvent):
+            e.hint.cols = None # type: ignore
         # Patch ModelChangedEvents which change an array property (see https://github.com/bokeh/bokeh/issues/11735)
         elif (isinstance(e, ModelChangedEvent) and isinstance(e.model, DataModel) and
               isinstance(e.new, np.ndarray)):
                 new_array = comparable_array(e.new.shape, e.new.dtype, e.new)
                 e.new = new_array
                 e.serializable_new = new_array
 
-
 #---------------------------------------------------------------------
 # Public API
 #---------------------------------------------------------------------
 
-
-
-
-def diff(doc, binary=True, events=None):
+def diff(
+    doc: 'Document', binary: bool = True, events: Optional[List['DocumentChangedEvent']] = None
+) -> Message[Any] | None:
     """
     Returns a json diff required to update an existing plot with
     the latest plot data.
@@ -68,13 +76,14 @@ def diff(doc, binary=True, events=None):
     if not events or state._hold:
         return None
 
-    patch_events(events)
-    msg = Protocol().create("PATCH-DOC", events, use_buffers=binary)
+    patch_events = [event for event in events if isinstance(event, DocumentPatchedEvent)]
+    monkeypatch_events(events)
+    msg_type: Literal["PATCH-DOC"] = "PATCH-DOC"
+    msg = Protocol().create(msg_type, patch_events, use_buffers=binary)
     doc.callbacks._held_events = [e for e in doc.callbacks._held_events if e not in events]
     return msg
 
-
-def remove_root(obj, replace=None):
+def remove_root(obj: 'Model', replace: Optional['Document'] = None) -> None:
     """
     Removes the document from any previously displayed bokeh object
     """
@@ -86,8 +95,7 @@ def remove_root(obj, replace=None):
         if replace:
             model._document = replace
 
-
-def add_to_doc(obj, doc, hold=False):
+def add_to_doc(obj: 'Model', doc: 'Document', hold: bool = False):
     """
     Adds a model to the supplied Document removing it from any existing Documents.
     """
@@ -97,9 +105,8 @@ def add_to_doc(obj, doc, hold=False):
     if doc.callbacks.hold_value is None and hold:
         doc.hold()
 
-
 @contextmanager
-def hold(doc, policy='combine', comm=None):
+def hold(doc: 'Document', policy: 'HoldPolicyType' = 'combine', comm: Optional['Comm'] = None):
     held = doc.callbacks.hold_value
     try:
         if policy is None:
@@ -116,7 +123,6 @@ def hold(doc, policy='combine', comm=None):
                 push(doc, comm)
             doc.unhold()
 
-
 def patch_cds_msg(model, msg):
     """
     Required for handling messages containing JSON serialized typed
@@ -132,10 +138,9 @@ def patch_cds_msg(model, msg):
             if isinstance(values, dict):
                 event['new'][col] = [v for _, v in sorted(values.items())]
 
-
 _DEFAULT_IGNORED_REPR = frozenset(['children', 'text', 'name', 'toolbar', 'renderers', 'below', 'center', 'left', 'right'])
 
-def bokeh_repr(obj, depth=0, ignored=None):
+def bokeh_repr(obj: 'Model', depth: int = 0, ignored: Optional[Iterable[str]] = None) -> str:
     """
     Returns a string repr for a bokeh model, useful for recreating
     panel objects using pure bokeh.
@@ -149,7 +154,7 @@ def bokeh_repr(obj, depth=0, ignored=None):
 
     r = ""
     cls = type(obj).__name__
-    properties = sorted(obj.properties_with_values(False).items())
+    properties = sorted(obj.properties_with_values(include_defaults=False).items())
     props = []
     for k, v in properties:
         if k in ignored:
@@ -161,12 +166,12 @@ def bokeh_repr(obj, depth=0, ignored=None):
         if len(v) > 30:
             v = v[:30] + '...'
         props.append('%s=%s' % (k, v))
-    props = ', '.join(props)
+    props_repr = ', '.join(props)
     if isinstance(obj, Box):
         r += '{cls}(children=[\n'.format(cls=cls)
-        for obj in obj.children:
+        for obj in obj.children: # type: ignore
             r += textwrap.indent(bokeh_repr(obj, depth=depth+1) + ',\n', '  ')
-        r += '], %s)' % props
+        r += '], %s)' % props_repr
     else:
-        r += '{cls}({props})'.format(cls=cls,  props=props)
+        r += '{cls}({props})'.format(cls=cls, props=props_repr)
     return r