refactor: ruff lint

src/jupynium/buffer.py

@@ -20,11 +20,12 @@ class JupyniumBuffer:
     Deal with the Nvim buffer and its cell information.
 
     This does have a functionality to sync with the Notebook.
+    It can also be part of the buffer, and anything above the first cell will be `header_cell_type`.
     """
 
     def __init__(
         self,
-        buf: list[str] = [""],
+        buf: list[str] | None = None,
         header_cell_type: str = "header",
     ):
         """
@@ -33,14 +34,15 @@ def __init__(
         Args:
             header_cell_type: Use only when partial update.
         """
-        self.buf = buf
-        if self.buf == [""]:
+        if buf is None:
             # each cell's row length. 0-th cell is not a cell, but it's the header.
             # You can put anything above and it won't be synced to Jupyter Notebook.
+            self.buf = [""]
             self.num_rows_per_cell: list[int] = [1]
 
             self.cell_types = ["header"]  # 0-th cell is not a cell.
         else:
+            self.buf = buf
             self.full_analyse_buf(header_cell_type)
 
     def full_analyse_buf(self, header_cell_type: str = "header"):
@@ -68,7 +70,7 @@ def full_analyse_buf(self, header_cell_type: str = "header"):
         num_rows_this_cell = 0
         num_rows_per_cell = []
         cell_types = [header_cell_type]
-        for row, line in enumerate(self.buf):
+        for _row, line in enumerate(self.buf):
             if line.startswith(("# %% [md]", "# %% [markdown]")):
                 num_rows_per_cell.append(num_rows_this_cell)
                 num_rows_this_cell = 1
@@ -178,7 +180,9 @@ def process_on_lines(
                 driver, modified_cell_idx_start, modified_cell_idx_end, strip=strip
             )
 
-    def _on_lines_update_buf(self, lines, start_row, old_end_row, new_end_row):
+    def _on_lines_update_buf(
+        self, lines: list[str], start_row: int, old_end_row: int, new_end_row: int
+    ):
         """Replace start_row:old_end_row to lines from self.buf."""
         # Analyse how many cells are removed
         notebook_cell_delete_operations = []
@@ -369,7 +373,7 @@ def _check_validity(self):
         assert all(x in ("code", "markdown") for x in self.cell_types[1:])
 
     def _partial_sync_to_notebook(
-        self, driver, start_cell_idx, end_cell_idx, strip=True
+        self, driver: WebDriver, start_cell_idx: int, end_cell_idx: int, strip=True
     ):
         """
         Given the range of cells to update, sync the JupyniumBuffer with the notebook.
@@ -437,7 +441,7 @@ def _partial_sync_to_notebook(
                 *texts_per_cell,
             )
 
-    def full_sync_to_notebook(self, driver, strip=True):
+    def full_sync_to_notebook(self, driver: WebDriver, strip: bool = True):
         # Full sync with notebook.
         # WARNING: syncing may result in data loss.
         num_cells = self.num_cells_in_notebook
@@ -473,7 +477,8 @@ def num_cells_in_notebook(self):
     def num_rows(self):
         return len(self.buf)
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
+        assert isinstance(other, JupyniumBuffer)
         return (
             self.buf == other.buf
             and self.num_rows_per_cell == other.num_rows_per_cell

src/jupynium/cmds/ipynb2jupytext.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import argparse
-import os
+from pathlib import Path
 
 from ..ipynb import ipynb2jupytext, load_ipynb
 
@@ -50,13 +50,14 @@ def main():
         for line in jupy:
             print(line)
     else:
-        output_jupy_path = args.output_jupy_path
-        if output_jupy_path is None:
-            output_jupy_path = os.path.splitext(args.ipynb_path)[0] + ".ju.py"
+        if args.output_jupy_path is None:
+            output_jupy_path = Path(args.ipynb_path).with_suffix(".ju.py")
+        else:
+            output_jupy_path = Path(args.output_jupy_path)
 
-        os.makedirs(os.path.dirname(os.path.realpath(output_jupy_path)), exist_ok=True)
+        output_jupy_path.parent.mkdir(parents=True, exist_ok=True)
 
-        if os.path.isfile(output_jupy_path) and not args.yes:
+        if output_jupy_path.is_file() and not args.yes:
             print(f"Do you want to overwrite {output_jupy_path}?")
             answer = input("y/n: ")
             if answer != "y":

src/jupynium/cmds/jupynium.py

@@ -413,7 +413,7 @@ def fallback_open_notebook_server(
         # e.g. --jupyter_command conda run ' --no-capture-output' ' -n' env_name jupyter
 
         jupyter_command = [command.strip() for command in jupyter_command]
-        jupyter_command[0] = os.path.expanduser(jupyter_command[0])
+        jupyter_command[0] = str(Path(jupyter_command[0]).expanduser())
 
         jupyter_stdout = tempfile.NamedTemporaryFile()
         logger.info(f"Writing Jupyter Notebook server log to: {jupyter_stdout.name}")

src/jupynium/events_control.py

@@ -6,6 +6,8 @@
 import os
 from dataclasses import dataclass
 from os import PathLike
+from pathlib import Path
+from typing import Any
 
 from pkg_resources import resource_stream
 from selenium.common.exceptions import (
@@ -130,7 +132,7 @@ class PrevLazyArgsPerBuf:
 
     data: dict[int, PrevLazyArgs] = dataclasses.field(default_factory=dict)
 
-    def process(self, bufnr: int, nvim_info: NvimInfo, driver) -> None:
+    def process(self, bufnr: int, nvim_info: NvimInfo, driver: WebDriver) -> None:
         if bufnr in self.data:
             self.data[bufnr].process(nvim_info, driver, bufnr)
 
@@ -140,7 +142,11 @@ def process_all(self, nvim_info: NvimInfo, driver) -> None:
         self.data.clear()
 
     def lazy_on_lines_event(
-        self, nvim_info: NvimInfo, driver, bufnr: int, on_lines_args: OnLinesArgs
+        self,
+        nvim_info: NvimInfo,
+        driver: WebDriver,
+        bufnr: int,
+        on_lines_args: OnLinesArgs,
     ) -> None:
         if bufnr not in self.data:
             self.data[bufnr] = PrevLazyArgs()
@@ -163,7 +169,7 @@ def overwrite_update_selection(
         self.data[bufnr].update_selection_args = update_selection_args
 
 
-def process_events(nvim_info: NvimInfo, driver):
+def process_events(nvim_info: NvimInfo, driver: WebDriver):
     """
     Controls events for a single nvim, and a single cycle of events.
 
@@ -241,6 +247,7 @@ def start_sync_with_filename(
     driver.switch_to.window(nvim_info.home_window)
     sele.wait_until_notebook_list_loaded(driver)
 
+    prev_windows = None
     if ipynb_filename == "":
         file_found = False
     else:
@@ -265,9 +272,11 @@ def start_sync_with_filename(
                 driver.execute_script("arguments[0].scrollIntoView();", notebook_elem)
                 notebook_elem.click()
                 file_found = True
-                sele.wait_until_new_window(driver, prev_windows)
+                sele.wait_until_new_window(driver, list(prev_windows))
                 break
 
+    assert prev_windows is not None
+
     if file_found:
         new_window = set(driver.window_handles) - set(prev_windows)
         assert len(new_window) == 1
@@ -313,7 +322,7 @@ def start_sync_with_filename(
             new_btn.click()
             kernel_btn.click()
 
-        sele.wait_until_new_window(driver, prev_windows)
+        sele.wait_until_new_window(driver, list(prev_windows))
         new_window = set(driver.window_handles) - prev_windows
         assert len(new_window) == 1
         new_window = new_window.pop()
@@ -333,9 +342,7 @@ def start_sync_with_filename(
 def choose_default_kernel(
     driver: WebDriver, page_type: str, buf_filetype: str, conda_or_venv_path: str | None
 ):
-    """
-    Choose kernel based on buffer's filetype and conda env
-    """
+    """Choose kernel based on buffer's filetype and conda env."""
     if page_type == "notebook":
         kernel_specs = driver.execute_script(
             "return Jupyter.kernelselector.kernelspecs;"
@@ -363,9 +370,9 @@ def match_with_path(env_path: str) -> str | None:
         """
         for kernel_name in valid_kernel_names:
             try:
-                kernel_exec_path = kernel_specs[kernel_name]["spec"]["argv"][0]
-                exec_name = os.path.basename(kernel_exec_path)
-                env_exec_path = os.path.join(env_path, "bin", exec_name)
+                kernel_exec_path = Path(kernel_specs[kernel_name]["spec"]["argv"][0])
+                exec_name = kernel_exec_path.name
+                env_exec_path = Path(env_path) / "bin" / exec_name
                 if kernel_exec_path == env_exec_path:
                     return kernel_name
             except (KeyError, IndexError):
@@ -413,8 +420,13 @@ def match_with_path(env_path: str) -> str | None:
     return None
 
 
-def process_request_event(nvim_info: NvimInfo, driver: WebDriver, event):
+def process_request_event(nvim_info: NvimInfo, driver: WebDriver, event: list[Any]):
     """
+    Process a request event, where an event can be request or notification.
+
+    Request event requires a response.
+    Notification event doesn't require a response.
+
     Returns:
         status (bool)
         request_event (rpcrequest event): to notify nvim after cleared up.
@@ -667,9 +679,9 @@ def process_notification_event(
                 ".ju." in buf_filepath
                 and nvim_info.nvim.vars["jupynium_auto_download_ipynb"]
             ):
-                output_ipynb_path = os.path.splitext(buf_filepath)[0]
-                output_ipynb_path = os.path.splitext(output_ipynb_path)[0]
-                output_ipynb_path += ".ipynb"
+                # .ju.py -> .ipynb
+                output_ipynb_path = os.path.splitext(buf_filepath)[0]  # noqa: PTH122
+                output_ipynb_path = Path(output_ipynb_path).with_suffix(".ipynb")
 
                 try:
                     download_ipynb(driver, nvim_info, bufnr, output_ipynb_path)
@@ -683,20 +695,21 @@ def process_notification_event(
             (buf_filepath, filename) = event_args
             assert buf_filepath != ""
 
+            buf_filepath = Path(buf_filepath)
+            filename = Path(filename)
+
             if filename is not None and filename != "":
-                if os.path.isabs(filename):
+                if filename.is_absolute():
                     output_ipynb_path = filename
                 else:
-                    output_ipynb_path = os.path.join(
-                        os.path.dirname(buf_filepath), filename
-                    )
+                    output_ipynb_path = buf_filepath.parent / filename
 
-                if not output_ipynb_path.endswith(".ipynb"):
-                    output_ipynb_path += ".ipynb"
+                if output_ipynb_path.suffix != ".ipynb":
+                    output_ipynb_path = output_ipynb_path.with_suffix(".ipynb")
             else:
-                output_ipynb_path = os.path.splitext(buf_filepath)[0]
-                output_ipynb_path = os.path.splitext(output_ipynb_path)[0]
-                output_ipynb_path += ".ipynb"
+                # change suffix .ju.py -> .ipynb
+                output_ipynb_path = os.path.splitext(buf_filepath)[0]  # noqa: PTH122
+                output_ipynb_path = Path(output_ipynb_path).with_suffix(".ipynb")
 
             try:
                 download_ipynb(driver, nvim_info, bufnr, output_ipynb_path)

src/jupynium/pynvim_helpers.py

@@ -2,14 +2,16 @@
 
 import logging
 import time
+from os import PathLike
 
 import pynvim
 from pkg_resources import resource_stream
 
 logger = logging.getLogger(__name__)
 
 
-def attach_and_init(nvim_listen_addr: str):
+def attach_and_init(nvim_listen_addr: str | PathLike):
+    nvim_listen_addr = str(nvim_listen_addr)
     logger.info("nvim addr: %s", nvim_listen_addr)
     for _ in range(30):
         try:

src/jupynium/rpc_messages.py

@@ -29,6 +29,7 @@ def receive_message(nvim: Nvim):
 def receive_all_pending_messages(nvim: Nvim):
     """
     It doesn't guarantee to grab all messages that are previously sent.
+
     Maybe the last one or two may still be in process.
     """
     events = []

tests/conftest.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-import os
 import subprocess
 import tempfile
+from pathlib import Path
 
 import pytest
 
@@ -18,7 +18,7 @@ def jupbuf1():
 @pytest.fixture(scope="session")
 def nvim_1():
     with tempfile.TemporaryDirectory() as tmp:
-        path = os.path.join(tmp, "nvim")
+        path = str(Path(tmp) / "nvim")
         nvim_proc = subprocess.Popen(
             ["nvim", "--clean", "--headless", "--listen", path]
         )