typehinting for clibuffer.py, several special .py

Add typehints for files:

 * mycli/clibuffer.py
 * mycli/packages/special/__init__.py
 * mycli/packages/special/dbcommands.py
 * mycli/packages/special/favoritequeries.py
 * mycli/packages/special/iocommands.py
 * mycli/packages/special/main.py

Changes
 * Import "iocommands" instead of toplevel "special"
 * create ArgType Enum for NO_QUERY and friends
 * convert special-command aliases argument to a list, since it is of
   variable length
 * rewrite open_external_editor to respect two different modes for
   invoking click.echo(): with and without filename
 * recast "log" as "logger" for consistency across the project
 * bugfix: always test whether fetchone() returns an unpackable value
 * don't set FavoriteQueries.instance to None
 * add some explicit returns
 * update docstrings
 * convert some format strings to f-strings
 * convert unassigned strings not in docstring position to comments
 * remove a Python 2 compatibility conditional
 * return a tuple from parseargfile() rather than a dict
 * reformat some long lines
 * remove "type: ignore"s
 * precede unused variables with underscores
 * don't rewrite variables with different-type values
 * add --non-interactive to CI when installing type stubs

Author: rolandwalker

.github/workflows/typecheck.yml

@@ -29,4 +29,5 @@ jobs:
 
       - name: Run mypy
         run: |
-          uv run --no-sync --frozen -- python -m mypy --no-pretty --install-types .
+          uv run --no-sync --frozen -- python -m ensurepip
+          uv run --no-sync --frozen -- python -m mypy --no-pretty --install-types --non-interactive .

mycli/clibuffer.py

@@ -1,13 +1,13 @@
-# type: ignore
+from typing import Callable
 
 from prompt_toolkit.application import get_app
 from prompt_toolkit.enums import DEFAULT_BUFFER
 from prompt_toolkit.filters import Condition
 
-from mycli.packages import special
+from mycli.packages.special import iocommands
 
 
-def cli_is_multiline(mycli):
+def cli_is_multiline(mycli) -> Callable:
     @Condition
     def cond():
         doc = get_app().layout.get_buffer_by_name(DEFAULT_BUFFER).document
@@ -20,7 +20,7 @@ def cond():
     return cond
 
 
-def _multiline_exception(text):
+def _multiline_exception(text: str) -> bool:
     orig = text
     text = text.strip()
 
@@ -39,7 +39,7 @@ def _multiline_exception(text):
         or
         # Ended with the current delimiter (usually a semi-column)
         text.endswith((
-            special.get_current_delimiter(),
+            iocommands.get_current_delimiter(),
             "\\g",
             "\\G",
             r"\e",

mycli/main.py

@@ -52,7 +52,7 @@
 from mycli.packages.parseutils import is_destructive, is_dropping_database
 from mycli.packages.prompt_utils import confirm, confirm_destructive_query
 from mycli.packages.special.favoritequeries import FavoriteQueries
-from mycli.packages.special.main import NO_QUERY
+from mycli.packages.special.main import ArgType
 from mycli.packages.tabular_output import sql_format
 from mycli.packages.toolkit.history import FileHistoryWithTimestamp
 from mycli.sqlcompleter import SQLCompleter
@@ -198,37 +198,37 @@ def __init__(
         self.prompt_app = None
 
     def register_special_commands(self):
-        special.register_special_command(self.change_db, "use", "\\u", "Change to a new database.", aliases=("\\u",))
+        special.register_special_command(self.change_db, "use", "\\u", "Change to a new database.", aliases=["\\u"])
         special.register_special_command(
             self.change_db,
             "connect",
             "\\r",
             "Reconnect to the database. Optional database argument.",
-            aliases=("\\r",),
+            aliases=["\\r"],
             case_sensitive=True,
         )
         special.register_special_command(
-            self.refresh_completions, "rehash", "\\#", "Refresh auto-completions.", arg_type=NO_QUERY, aliases=("\\#",)
+            self.refresh_completions, "rehash", "\\#", "Refresh auto-completions.", arg_type=ArgType.NO_QUERY, aliases=["\\#"]
         )
         special.register_special_command(
             self.change_table_format,
             "tableformat",
             "\\T",
             "Change the table format used to output results.",
-            aliases=("\\T",),
+            aliases=["\\T"],
             case_sensitive=True,
         )
         special.register_special_command(
             self.change_redirect_format,
             "redirectformat",
             "\\Tr",
             "Change the table format used to output redirected results.",
-            aliases=("\\Tr",),
+            aliases=["\\Tr"],
             case_sensitive=True,
         )
-        special.register_special_command(self.execute_from_file, "source", "\\. filename", "Execute commands from file.", aliases=("\\.",))
+        special.register_special_command(self.execute_from_file, "source", "\\. filename", "Execute commands from file.", aliases=["\\."])
         special.register_special_command(
-            self.change_prompt_format, "prompt", "\\R", "Change prompt format.", aliases=("\\R",), case_sensitive=True
+            self.change_prompt_format, "prompt", "\\R", "Change prompt format.", aliases=["\\R"], case_sensitive=True
         )
 
     def change_table_format(self, arg, **_):
@@ -574,7 +574,7 @@ def handle_editor_command(self, text):
         while special.editor_command(text):
             filename = special.get_filename(text)
             query = special.get_editor_query(text) or self.get_last_query()
-            sql, message = special.open_external_editor(filename, sql=query)
+            sql, message = special.open_external_editor(filename=filename, sql=query)
             if message:
                 # Something went wrong. Raise an exception and bail.
                 raise RuntimeError(message)

mycli/packages/parseutils.py

@@ -123,7 +123,7 @@ def extract_from_part(parsed: TokenList, stop_at_punctuation: bool = True) -> Ge
                     break
 
 
-def extract_table_identifiers(token_stream: TokenList) -> Generator[tuple[str | None, str, str]]:
+def extract_table_identifiers(token_stream: TokenList) -> Generator[tuple[str | None, str, str], None, None]:
     """yields tuples of (schema_name, table_name, table_alias)"""
 
     for item in token_stream:

mycli/packages/special/__init__.py

@@ -1,9 +1,11 @@
-# type: ignore
+from __future__ import annotations
 
-__all__ = []
+from typing import Callable
 
+__all__: list[str] = []
 
-def export(defn):
+
+def export(defn: Callable):
     """Decorator to explicitly mark functions that are exposed in a lib."""
     globals()[defn.__name__] = defn
     __all__.append(defn.__name__)

mycli/packages/special/dbcommands.py

@@ -1,26 +1,32 @@
-# type: ignore
+from __future__ import annotations
 
 import logging
 import os
 import platform
 
 from pymysql import ProgrammingError
+from pymysql.cursors import Cursor
 
 from mycli import __version__
 from mycli.packages.special import iocommands
-from mycli.packages.special.main import PARSED_QUERY, RAW_QUERY, special_command
+from mycli.packages.special.main import ArgType, special_command
 from mycli.packages.special.utils import format_uptime
 
-log = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
-@special_command("\\dt", "\\dt[+] [table]", "List or describe tables.", arg_type=PARSED_QUERY, case_sensitive=True)
-def list_tables(cur, arg=None, arg_type=PARSED_QUERY, verbose=False):
+@special_command("\\dt", "\\dt[+] [table]", "List or describe tables.", arg_type=ArgType.PARSED_QUERY, case_sensitive=True)
+def list_tables(
+    cur: Cursor,
+    arg: str | None = None,
+    _arg_type: ArgType = ArgType.PARSED_QUERY,
+    verbose: bool = False,
+) -> list[tuple]:
     if arg:
         query = "SHOW FIELDS FROM {0}".format(arg)
     else:
         query = "SHOW TABLES"
-    log.debug(query)
+    logger.debug(query)
     cur.execute(query)
     tables = cur.fetchall()
     status = ""
@@ -31,17 +37,18 @@ def list_tables(cur, arg=None, arg_type=PARSED_QUERY, verbose=False):
 
     if verbose and arg:
         query = "SHOW CREATE TABLE {0}".format(arg)
-        log.debug(query)
+        logger.debug(query)
         cur.execute(query)
-        status = cur.fetchone()[1]
+        if one := cur.fetchone():
+            status = one[1]
 
     return [(None, tables, headers, status)]
 
 
-@special_command("\\l", "\\l", "List databases.", arg_type=RAW_QUERY, case_sensitive=True)
-def list_databases(cur, **_):
+@special_command("\\l", "\\l", "List databases.", arg_type=ArgType.RAW_QUERY, case_sensitive=True)
+def list_databases(cur: Cursor, **_) -> list[tuple]:
     query = "SHOW DATABASES"
-    log.debug(query)
+    logger.debug(query)
     cur.execute(query)
     if cur.description:
         headers = [x[0] for x in cur.description]
@@ -50,21 +57,23 @@ def list_databases(cur, **_):
         return [(None, None, None, "")]
 
 
-@special_command("status", "\\s", "Get status information from the server.", arg_type=RAW_QUERY, aliases=("\\s",), case_sensitive=True)
-def status(cur, **_):
+@special_command(
+    "status", "\\s", "Get status information from the server.", arg_type=ArgType.RAW_QUERY, aliases=["\\s"], case_sensitive=True
+)
+def status(cur: Cursor, **_) -> list[tuple]:
     query = "SHOW GLOBAL STATUS;"
-    log.debug(query)
+    logger.debug(query)
     try:
         cur.execute(query)
     except ProgrammingError:
         # Fallback in case query fail, as it does with Mysql 4
         query = "SHOW STATUS;"
-        log.debug(query)
+        logger.debug(query)
         cur.execute(query)
     status = dict(cur.fetchall())
 
     query = "SHOW GLOBAL VARIABLES;"
-    log.debug(query)
+    logger.debug(query)
     cur.execute(query)
     variables = dict(cur.fetchall())
 
@@ -92,11 +101,13 @@ def status(cur, **_):
     output.append(("Connection id:", cur.connection.thread_id()))
 
     query = "SELECT DATABASE(), USER();"
-    log.debug(query)
+    logger.debug(query)
     cur.execute(query)
-    db, user = cur.fetchone()
-    if db is None:
+    if one := cur.fetchone():
+        db, user = one
+    else:
         db = ""
+        user = ""
 
     output.append(("Current database:", db))
     output.append(("Current user:", user))
@@ -121,9 +132,12 @@ def status(cur, **_):
     output.append(("Connection:", host_info))
 
     query = "SELECT @@character_set_server, @@character_set_database, @@character_set_client, @@character_set_connection LIMIT 1;"
-    log.debug(query)
+    logger.debug(query)
     cur.execute(query)
-    charset = cur.fetchone()
+    if one := cur.fetchone():
+        charset = one
+    else:
+        charset = ("", "", "", "")
     output.append(("Server characterset:", charset[0]))
     output.append(("Db characterset:", charset[1]))
     output.append(("Client characterset:", charset[2]))
@@ -151,8 +165,8 @@ def status(cur, **_):
         if "Queries" in status:
             queries_per_second = int(status["Queries"]) / int(status["Uptime"])
             stats.append("Queries per second avg: {:.3f}".format(queries_per_second))
-        stats = "  ".join(stats)
-        footer.append("\n" + stats)
+        stats_str = "  ".join(stats)
+        footer.append("\n" + stats_str)
 
     footer.append("--------------")
     return [("\n".join(title), output, "", "\n".join(footer))]

mycli/packages/special/favoritequeries.py

@@ -34,7 +34,7 @@ class FavoriteQueries:
 """
 
     # Class-level variable, for convenience to use as a singleton.
-    instance = None
+    instance: FavoriteQueries
 
     def __init__(self, config) -> None:
         self.config = config