Refactoring and tests for plugin Host and decorators

pynvim/__init__.py

@@ -44,6 +44,8 @@ def start_host(session: Optional[Session] = None) -> None:
     This function is normally called at program startup and could have been
     defined as a separate executable. It is exposed as a library function for
     testing purposes only.
+
+    See also $VIMRUNTIME/autoload/provider/pythonx.vim for python host startup.
     """
     plugins = []
     for arg in sys.argv:

pynvim/api/nvim.py

@@ -24,10 +24,6 @@
 if TYPE_CHECKING:
     from pynvim.msgpack_rpc import Session
 
-if sys.version_info < (3, 8):
-    from typing_extensions import Literal
-else:
-    from typing import Literal
 
 __all__ = ['Nvim']
 
@@ -281,7 +277,7 @@ def __exit__(self, *exc_info: Any) -> None:
         """
         self.close()
 
-    def with_decode(self, decode: Literal[True] = True) -> Nvim:
+    def with_decode(self, decode: TDecodeMode = True) -> Nvim:
         """Initialize a new Nvim instance."""
         return Nvim(self._session, self.channel_id,
                     self.metadata, self.types, decode, self._err_cb)

pynvim/plugin/__init__.py

@@ -2,7 +2,7 @@
 
 from pynvim.plugin.decorators import (autocmd, command, decode, encoding, function,
                                       plugin, rpc_export, shutdown_hook)
-from pynvim.plugin.host import Host  # type: ignore[attr-defined]
+from pynvim.plugin.host import Host
 
 
 __all__ = ('Host', 'plugin', 'rpc_export', 'command', 'autocmd',

pynvim/plugin/decorators.py

@@ -1,24 +1,89 @@
 """Decorators used by python host plugin system."""
 
+from __future__ import annotations
+
 import inspect
 import logging
 import sys
-from typing import Any, Callable, Dict, Optional, TypeVar, Union
-
-from pynvim.compat import unicode_errors_default
+from functools import partial
+from typing import (TYPE_CHECKING, Any, Callable, Dict, Optional, Type,
+                    TypeVar, Union, cast, overload)
 
 if sys.version_info < (3, 8):
-    from typing_extensions import Literal
+    from typing_extensions import Literal, Protocol, TypedDict
 else:
-    from typing import Literal
+    from typing import Literal, Protocol, TypedDict
+
+if sys.version_info < (3, 10):
+    from typing_extensions import ParamSpec
+else:
+    from typing import ParamSpec
+
+from pynvim.api.common import TDecodeMode
+from pynvim.compat import unicode_errors_default
 
 logger = logging.getLogger(__name__)
-debug, info, warn = (logger.debug, logger.info, logger.warning,)
+debug, info, warn = (
+    logger.debug,
+    logger.info,
+    logger.warning,
+)
 __all__ = ('plugin', 'rpc_export', 'command', 'autocmd', 'function',
            'encoding', 'decode', 'shutdown_hook')
 
 T = TypeVar('T')
-F = TypeVar('F', bound=Callable[..., Any])
+
+if TYPE_CHECKING:
+
+    class RpcSpec(TypedDict):
+        type: Literal['command', 'autocmd', 'function']
+        name: str
+        sync: Union[bool, Literal['urgent']]
+        opts: Any
+else:
+    RpcSpec = dict
+
+# type variables for Handler, to represent Callable: P -> R
+P = ParamSpec('P')
+R = TypeVar('R')
+
+
+class Handler(Protocol[P, R]):
+    """An interface to pynvim-decorated RPC handler.
+
+    Handler is basically a callable (method) that is decorated by pynvim.
+    It will have some private fields (prefixed with `_nvim_`), set by
+    decorators that follow below. This generic type allows stronger, static
+    typing for all the private attributes (see `host.Host` for the usage).
+
+    Note: Any valid Handler that is created by pynvim's decorator is guaranteed
+    to have *all* of the following `_nvim_*` attributes defined as per the
+    "Protocol", so there is NO need to check `hasattr(handler, "_nvim_...")`.
+    Exception is _nvim_decode; this is an optional attribute orthgonally set by
+    the decorator `@decode()`.
+    """
+    __call__: Callable[P, R]
+
+    _nvim_rpc_method_name: str
+    _nvim_rpc_sync: bool
+    _nvim_bind: bool
+    _nvim_prefix_plugin_path: bool
+    _nvim_rpc_spec: Optional[RpcSpec]
+    _nvim_shutdown_hook: bool
+
+    _nvim_registered_name: Optional[str]  # set later by host when discovered
+
+    @classmethod
+    def wrap(cls, fn: Callable[P, R]) -> Handler[P, R]:
+        fn = cast(Handler[P, R], partial(fn))
+        fn._nvim_bind = False
+        fn._nvim_rpc_method_name = None  # type: ignore
+        fn._nvim_rpc_sync = None  # type: ignore
+        fn._nvim_prefix_plugin_path = False
+        fn._nvim_rpc_spec = None
+        fn._nvim_shutdown_hook = False
+        fn._nvim_registered_name = None
+        return fn
 
 
 def plugin(cls: T) -> T:
@@ -28,24 +93,34 @@ def plugin(cls: T) -> T:
     plugin_load method of the host.
     """
     cls._nvim_plugin = True  # type: ignore[attr-defined]
+
     # the _nvim_bind attribute is set to True by default, meaning that
     # decorated functions have a bound Nvim instance as first argument.
     # For methods in a plugin-decorated class this is not required, because
     # the class initializer will already receive the nvim object.
-    predicate = lambda fn: hasattr(fn, '_nvim_bind')
+    predicate = lambda fn: getattr(fn, '_nvim_bind', False)
     for _, fn in inspect.getmembers(cls, predicate):
         fn._nvim_bind = False
     return cls
 
 
-def rpc_export(rpc_method_name: str, sync: bool = False) -> Callable[[F], F]:
+def rpc_export(
+    rpc_method_name: str,
+    sync: bool = False,
+) -> Callable[[Callable[P, R]], Handler[P, R]]:
     """Export a function or plugin method as a msgpack-rpc request handler."""
-    def dec(f: F) -> F:
-        f._nvim_rpc_method_name = rpc_method_name  # type: ignore[attr-defined]
-        f._nvim_rpc_sync = sync  # type: ignore[attr-defined]
-        f._nvim_bind = True  # type: ignore[attr-defined]
-        f._nvim_prefix_plugin_path = False  # type: ignore[attr-defined]
+
+    def dec(f: Callable[P, R]) -> Handler[P, R]:
+        f = cast(Handler[P, R], f)
+        f._nvim_rpc_method_name = rpc_method_name
+        f._nvim_rpc_sync = sync
+        f._nvim_bind = True
+        f._nvim_prefix_plugin_path = False
+        f._nvim_rpc_spec = None  # not used
+        f._nvim_shutdown_hook = False  # not used
+        f._nvim_registered_name = None  # TBD
         return f
+
     return dec
 
 
@@ -60,15 +135,15 @@ def command(
     sync: bool = False,
     allow_nested: bool = False,
     eval: Optional[str] = None
-) -> Callable[[F], F]:
+) -> Callable[[Callable[P, R]], Handler[P, R]]:
     """Tag a function or plugin method as a Nvim command handler."""
-    def dec(f: F) -> F:
-        f._nvim_rpc_method_name = (  # type: ignore[attr-defined]
-            'command:{}'.format(name)
-        )
-        f._nvim_rpc_sync = sync  # type: ignore[attr-defined]
-        f._nvim_bind = True  # type: ignore[attr-defined]
-        f._nvim_prefix_plugin_path = True  # type: ignore[attr-defined]
+
+    def dec(f: Callable[P, R]) -> Handler[P, R]:
+        f = cast(Handler[P, R], f)
+        f._nvim_rpc_method_name = ('command:{}'.format(name))
+        f._nvim_rpc_sync = sync
+        f._nvim_bind = True
+        f._nvim_prefix_plugin_path = True
 
         opts: Dict[str, Any] = {}
 
@@ -97,13 +172,16 @@ def dec(f: F) -> F:
         else:
             rpc_sync = sync
 
-        f._nvim_rpc_spec = {  # type: ignore[attr-defined]
+        f._nvim_rpc_spec = {
             'type': 'command',
             'name': name,
             'sync': rpc_sync,
             'opts': opts
         }
+        f._nvim_shutdown_hook = False
+        f._nvim_registered_name = None  # TBD
         return f
+
     return dec
 
 
@@ -113,19 +191,17 @@ def autocmd(
     sync: bool = False,
     allow_nested: bool = False,
     eval: Optional[str] = None
-) -> Callable[[F], F]:
+) -> Callable[[Callable[P, R]], Handler[P, R]]:
     """Tag a function or plugin method as a Nvim autocommand handler."""
-    def dec(f: F) -> F:
-        f._nvim_rpc_method_name = (  # type: ignore[attr-defined]
-            'autocmd:{}:{}'.format(name, pattern)
-        )
-        f._nvim_rpc_sync = sync  # type: ignore[attr-defined]
-        f._nvim_bind = True  # type: ignore[attr-defined]
-        f._nvim_prefix_plugin_path = True  # type: ignore[attr-defined]
-
-        opts = {
-            'pattern': pattern
-        }
+
+    def dec(f: Callable[P, R]) -> Handler[P, R]:
+        f = cast(Handler[P, R], f)
+        f._nvim_rpc_method_name = ('autocmd:{}:{}'.format(name, pattern))
+        f._nvim_rpc_sync = sync
+        f._nvim_bind = True
+        f._nvim_prefix_plugin_path = True
+
+        opts = {'pattern': pattern}
 
         if eval:
             opts['eval'] = eval
@@ -135,13 +211,16 @@ def dec(f: F) -> F:
         else:
             rpc_sync = sync
 
-        f._nvim_rpc_spec = {  # type: ignore[attr-defined]
+        f._nvim_rpc_spec = {
             'type': 'autocmd',
             'name': name,
             'sync': rpc_sync,
             'opts': opts
         }
+        f._nvim_shutdown_hook = False
+        f._nvim_registered_name = None  # TBD
         return f
+
     return dec
 
 
@@ -151,15 +230,15 @@ def function(
     sync: bool = False,
     allow_nested: bool = False,
     eval: Optional[str] = None
-) -> Callable[[F], F]:
+) -> Callable[[Callable[P, R]], Handler[P, R]]:
     """Tag a function or plugin method as a Nvim function handler."""
-    def dec(f: F) -> F:
-        f._nvim_rpc_method_name = (  # type: ignore[attr-defined]
-            'function:{}'.format(name)
-        )
-        f._nvim_rpc_sync = sync  # type: ignore[attr-defined]
-        f._nvim_bind = True  # type: ignore[attr-defined]
-        f._nvim_prefix_plugin_path = True  # type: ignore[attr-defined]
+
+    def dec(f: Callable[P, R]) -> Handler[P, R]:
+        f = cast(Handler[P, R], f)
+        f._nvim_rpc_method_name = ('function:{}'.format(name))
+        f._nvim_rpc_sync = sync
+        f._nvim_bind = True
+        f._nvim_prefix_plugin_path = True
 
         opts = {}
 
@@ -174,37 +253,79 @@ def dec(f: F) -> F:
         else:
             rpc_sync = sync
 
-        f._nvim_rpc_spec = {  # type: ignore[attr-defined]
+        f._nvim_rpc_spec = {
             'type': 'function',
             'name': name,
             'sync': rpc_sync,
             'opts': opts
         }
+        f._nvim_shutdown_hook = False  # not used
+        f._nvim_registered_name = None  # TBD
         return f
+
     return dec
 
 
-def shutdown_hook(f: F) -> F:
+def shutdown_hook(f: Callable[P, R]) -> Handler[P, R]:
     """Tag a function or method as a shutdown hook."""
-    f._nvim_shutdown_hook = True  # type: ignore[attr-defined]
-    f._nvim_bind = True  # type: ignore[attr-defined]
+    f = cast(Handler[P, R], f)
+    f._nvim_rpc_method_name = ''  # Falsy value, not used
+    f._nvim_rpc_sync = True  # not used
+    f._nvim_prefix_plugin_path = False  # not used
+    f._nvim_rpc_spec = None  # not used
+
+    f._nvim_shutdown_hook = True
+    f._nvim_bind = True
+    f._nvim_registered_name = None  # TBD
     return f
 
 
-def decode(mode: str = unicode_errors_default) -> Callable[[F], F]:
-    """Configure automatic encoding/decoding of strings."""
-    def dec(f: F) -> F:
-        f._nvim_decode = mode  # type: ignore[attr-defined]
+T_Decode = Union[Type, Handler[P, R]]
+
+
+def decode(
+    mode: TDecodeMode = unicode_errors_default,
+) -> Callable[[T_Decode], T_Decode]:
+    """Configure automatic encoding/decoding of strings.
+
+    This decorator can be put around an individual Handler (@rpc_export,
+    @autocmd, @function, @command, or @shutdown_hook), or around a class
+    (@plugin, has an effect on all the methods unless overridden).
+
+    The argument `mode` will be passed as an argument to:
+        bytes.decode("utf-8", errors=mode)
+    when decoding bytestream Nvim RPC responses.
+
+    See https://docs.python.org/3/library/codecs.html#error-handlers for
+    the list of valid modes (error handler values).
+
+    See also:
+        pynvim.api.Nvim.with_decode(mode)
+        pynvim.api.common.decode_if_bytes(..., mode)
+    """
+
+    @overload
+    def dec(f: Handler[P, R]) -> Handler[P, R]:
+        ...  # decorator on method
+
+    @overload
+    def dec(f: Type[T]) -> Type[T]:
+        ...  # decorator on class
+
+    def dec(f):  # type: ignore
+        f._nvim_decode = mode
         return f
-    return dec
 
+    return dec  # type: ignore
 
-def encoding(encoding: Union[bool, str] = True) -> Callable[[F], F]:
+
+def encoding(encoding: Union[bool, str] = True):  # type: ignore
     """DEPRECATED: use pynvim.decode()."""
     if isinstance(encoding, str):
         encoding = True
 
-    def dec(f: F) -> F:
-        f._nvim_decode = encoding  # type: ignore[attr-defined]
+    def dec(f):  # type: ignore
+        f._nvim_decode = encoding if encoding else None
         return f
+
     return dec