typecheck for several files including filepaths.py

add type annotations for
 * mycli/__init__.py
 * mycli/compat.py
 * mycli/lexer.py
 * mycli/packages/filepaths.py
 * mycli/packages/hybrid_redirection.py
 * mycli/packages/parseutils.py
 * mycli/packages/special/delimitercommand.py
 * mycli/packages/toolkit/fzf.py
 * mycli/packages/toolkit/history.py
 * mycli/packages/special/favoritequeries.py
 * mycli/packages/special/utils.py
 * mycli/packages/tabular_output/sql_format.py

and enable mypy checking for them in CI.

Convert capitalized types in toolkit/history.py to modern annotations.

Fix an int() cast on an env variable in test/utils.py, without enabling
typecheck for the entire file.

Make query_starts_with() and queries_starts_with() take a list rather
than a tuple for the second argument.

Fix a bug where extract_tables_from_complete_statements could crash
without checking for a None.

Fix a bug where table_format was not checked for None before treating
as a string.

Move __main__ section of parseutils.py to bottom of file.  Consider
removing it.

Remove a needless inherit from "object".

Convert an if/elif to the guard clause pattern.

Author: rolandwalker

changelog.md

@@ -5,6 +5,7 @@ Internal
 --------
 
 * Add limited typechecking to CI.
+* Enable typechecking for several files.
 
 
 1.35.0 (2025/07/18)

mycli/__init__.py

@@ -1,5 +1,5 @@
-# type: ignore
+from __future__ import annotations
 
 import importlib.metadata
 
-__version__ = importlib.metadata.version("mycli")
+__version__: str = importlib.metadata.version("mycli")

mycli/compat.py

@@ -1,7 +1,7 @@
-# type: ignore
-
 """Platform and Python version compatibility support."""
 
+from __future__ import annotations
+
 import sys
 
-WIN = sys.platform in ("win32", "cygwin")
+WIN: bool = sys.platform in ("win32", "cygwin")

mycli/lexer.py

@@ -1,4 +1,4 @@
-# type: ignore
+from __future__ import annotations
 
 from pygments.lexer import inherit
 from pygments.lexers.sql import MySqlLexer

mycli/packages/filepaths.py

@@ -1,18 +1,17 @@
-# type: ignore
+from __future__ import annotations
 
 import os
 import platform
 
+DEFAULT_SOCKET_DIRS: list[str] = []
 if os.name == "posix":
     if platform.system() == "Darwin":
         DEFAULT_SOCKET_DIRS = ["/tmp"]
     else:
         DEFAULT_SOCKET_DIRS = ["/var/run", "/var/lib"]
-else:
-    DEFAULT_SOCKET_DIRS = []
 
 
-def list_path(root_dir):
+def list_path(root_dir: str) -> list[str]:
     """List directory if exists.
 
     :param root_dir: str
@@ -26,7 +25,7 @@ def list_path(root_dir):
     return res
 
 
-def complete_path(curr_dir, last_dir):
+def complete_path(curr_dir: str, last_dir: str) -> str:
     """Return the path to complete that matches the last entered component.
 
     If the last entered component is ~, expanded path would not
@@ -41,9 +40,11 @@ def complete_path(curr_dir, last_dir):
         return curr_dir
     elif last_dir == "~":
         return os.path.join(last_dir, curr_dir)
+    else:
+        return ''
 
 
-def parse_path(root_dir):
+def parse_path(root_dir: str) -> tuple[str, str, int]:
     """Split path into head and last component for the completer.
 
     Also return position where last component starts.
@@ -59,7 +60,7 @@ def parse_path(root_dir):
     return base_dir, last_dir, position
 
 
-def suggest_path(root_dir):
+def suggest_path(root_dir: str) -> list[str]:
     """List all files and subdirectories in a directory.
 
     If the directory is not specified, suggest root directory,
@@ -81,7 +82,7 @@ def suggest_path(root_dir):
     return list_path(root_dir)
 
 
-def dir_path_exists(path):
+def dir_path_exists(path: str) -> bool:
     """Check if the directory path exists for a given file.
 
     For example, for a file /home/user/.cache/mycli/log, check if
@@ -94,7 +95,7 @@ def dir_path_exists(path):
     return os.path.exists(os.path.dirname(path))
 
 
-def guess_socket_location():
+def guess_socket_location() -> str | None:
     """Try to guess the location of the default mysql socket file."""
     socket_dirs = filter(os.path.exists, DEFAULT_SOCKET_DIRS)
     for directory in socket_dirs:

mycli/packages/hybrid_redirection.py

@@ -5,8 +5,8 @@
 
 import sqlglot
 
-from mycli.compat import WIN  # type: ignore[attr-defined]
-from mycli.packages.special.delimitercommand import DelimiterCommand  # type: ignore[attr-defined]
+from mycli.compat import WIN
+from mycli.packages.special.delimitercommand import DelimiterCommand
 
 logger = logging.getLogger(__name__)
 delimiter_command = DelimiterCommand()

mycli/packages/parseutils.py

@@ -1,13 +1,14 @@
-# type: ignore
+from __future__ import annotations
 
 import re
+from typing import Generator
 
 import sqlglot
-import sqlparse  # type: ignore[import-untyped]
-from sqlparse.sql import Function, Identifier, IdentifierList  # type: ignore[import-untyped]
-from sqlparse.tokens import DML, Keyword, Punctuation  # type: ignore[import-untyped]
+import sqlparse
+from sqlparse.sql import Function, Identifier, IdentifierList, Token, TokenList
+from sqlparse.tokens import DML, Keyword, Punctuation
 
-cleanup_regex = {
+cleanup_regex: dict[str, re.Pattern] = {
     # This matches only alphanumerics and underscores.
     "alphanum_underscore": re.compile(r"(\w+)$"),
     # This matches everything except spaces, parens, colon, and comma
@@ -19,7 +20,7 @@
 }
 
 
-def last_word(text, include="alphanum_underscore"):
+def last_word(text: str, include: str = "alphanum_underscore") -> str:
     r"""
     Find the last word in a sentence.
 
@@ -67,7 +68,7 @@ def last_word(text, include="alphanum_underscore"):
 
 # This code is borrowed from sqlparse example script.
 # <url>
-def is_subselect(parsed):
+def is_subselect(parsed: TokenList) -> bool:
     if not parsed.is_group:
         return False
     for item in parsed.tokens:
@@ -76,15 +77,15 @@ def is_subselect(parsed):
     return False
 
 
-def extract_from_part(parsed, stop_at_punctuation=True):
+def extract_from_part(parsed: TokenList, stop_at_punctuation: bool = True) -> Generator[str, None, None]:
     tbl_prefix_seen = False
     for item in parsed.tokens:
         if tbl_prefix_seen:
             if is_subselect(item):
                 for x in extract_from_part(item, stop_at_punctuation):
                     yield x
             elif stop_at_punctuation and item.ttype is Punctuation:
-                return
+                return None
             # Multiple JOINs in the same query won't work properly since
             # "ON" is a keyword and will trigger the next elif condition.
             # So instead of stooping the loop when finding an "ON" skip it
@@ -101,7 +102,7 @@ def extract_from_part(parsed, stop_at_punctuation=True):
             # condition. So we need to ignore the keyword JOIN and its variants
             # INNER JOIN, FULL OUTER JOIN, etc.
             elif item.ttype is Keyword and (not item.value.upper() == "FROM") and (not item.value.upper().endswith("JOIN")):
-                return
+                return None
             else:
                 yield item
         elif (item.ttype is Keyword or item.ttype is Keyword.DML) and item.value.upper() in (
@@ -122,7 +123,7 @@ def extract_from_part(parsed, stop_at_punctuation=True):
                     break
 
 
-def extract_table_identifiers(token_stream):
+def extract_table_identifiers(token_stream: TokenList) -> Generator[tuple[str | None, str, str]]:
     """yields tuples of (schema_name, table_name, table_alias)"""
 
     for item in token_stream:
@@ -151,7 +152,7 @@ def extract_table_identifiers(token_stream):
 
 
 # extract_tables is inspired from examples in the sqlparse lib.
-def extract_tables(sql):
+def extract_tables(sql: str) -> list[tuple[str | None, str, str]]:
     """Extract the table names from an SQL statement.
 
     Returns a list of (schema, table, alias) tuples
@@ -170,7 +171,7 @@ def extract_tables(sql):
     return list(extract_table_identifiers(stream))
 
 
-def extract_tables_from_complete_statements(sql):
+def extract_tables_from_complete_statements(sql: str) -> list[tuple[str | None, str, str | None]]:
     """Extract the table names from a complete and valid series of SQL
     statements.
 
@@ -195,7 +196,7 @@ def extract_tables_from_complete_statements(sql):
     tables = []
     for statement in finely_parsed:
         for identifier in statement.find_all(sqlglot.exp.Table):
-            if identifier.parent_select.sql().startswith('WITH'):
+            if identifier.parent_select and identifier.parent_select.sql().startswith('WITH'):
                 continue
             tables.append((
                 None if identifier.db == '' else identifier.db,
@@ -206,7 +207,7 @@ def extract_tables_from_complete_statements(sql):
     return tables
 
 
-def find_prev_keyword(sql):
+def find_prev_keyword(sql: str) -> tuple[Token | None, str]:
     """Find the last sql keyword in an SQL statement
 
     Returns the value of the last keyword, and the text of the query with
@@ -240,45 +241,40 @@ def find_prev_keyword(sql):
     return None, ""
 
 
-def query_starts_with(query, prefixes):
+def query_starts_with(query: str, prefixes: list[str]) -> bool:
     """Check if the query starts with any item from *prefixes*."""
     prefixes = [prefix.lower() for prefix in prefixes]
     formatted_sql = sqlparse.format(query.lower(), strip_comments=True)
     return bool(formatted_sql) and formatted_sql.split()[0] in prefixes
 
 
-def queries_start_with(queries, prefixes):
+def queries_start_with(queries: str, prefixes: list[str]) -> bool:
     """Check if any queries start with any item from *prefixes*."""
     for query in sqlparse.split(queries):
         if query and query_starts_with(query, prefixes) is True:
             return True
     return False
 
 
-def query_has_where_clause(query):
+def query_has_where_clause(query: str) -> bool:
     """Check if the query contains a where-clause."""
     return any(isinstance(token, sqlparse.sql.Where) for token_list in sqlparse.parse(query) for token in token_list)
 
 
-def is_destructive(queries):
+def is_destructive(queries: str) -> bool:
     """Returns if any of the queries in *queries* is destructive."""
     keywords = ("drop", "shutdown", "delete", "truncate", "alter")
     for query in sqlparse.split(queries):
         if query:
-            if query_starts_with(query, keywords) is True:
+            if query_starts_with(query, list(keywords)) is True:
                 return True
             elif query_starts_with(query, ["update"]) is True and not query_has_where_clause(query):
                 return True
 
     return False
 
 
-if __name__ == "__main__":
-    sql = "select * from (select t. from tabl t"
-    print(extract_tables(sql))
-
-
-def is_dropping_database(queries, dbname):
+def is_dropping_database(queries: list[str], dbname: str | None) -> bool:
     """Determine if the query is dropping a specific database."""
     result = False
     if dbname is None:
@@ -301,3 +297,8 @@ def normalize_db_name(db):
             if database_token is not None and normalize_db_name(database_token.get_name()) == dbname:
                 result = keywords[0].normalized == "DROP"
     return result
+
+
+if __name__ == "__main__":
+    sql = "select * from (select t. from tabl t"
+    print(extract_tables(sql))

mycli/packages/special/delimitercommand.py

@@ -1,15 +1,16 @@
-# type: ignore
+from __future__ import annotations
 
 import re
+from typing import Generator
 
 import sqlparse  # type: ignore[import-untyped]
 
 
 class DelimiterCommand(object):
-    def __init__(self):
+    def __init__(self) -> None:
         self._delimiter = ";"
 
-    def _split(self, sql):
+    def _split(self, sql: str) -> list[str]:
         """Temporary workaround until sqlparse.split() learns about custom
         delimiters."""
 
@@ -29,7 +30,7 @@ def _split(self, sql):
 
         return [stmt.replace(";", self._delimiter).replace(placeholder, ";") for stmt in split]
 
-    def queries_iter(self, input_str):
+    def queries_iter(self, input_str: str) -> Generator[str, None, None]:
         """Iterate over queries in the input string."""
 
         queries = self._split(input_str)
@@ -54,7 +55,7 @@ def queries_iter(self, input_str):
                         combined_statement += delimiter
                     queries = self._split(combined_statement)[1:]
 
-    def set(self, arg, **_):
+    def set(self, arg: str, **_) -> list[tuple[None, None, None, str]]:
         """Change delimiter.
 
         Since `arg` is everything that follows the DELIMITER token
@@ -76,5 +77,5 @@ def set(self, arg, **_):
         return [(None, None, None, "Changed delimiter to {}".format(delimiter))]
 
     @property
-    def current(self):
+    def current(self) -> str:
         return self._delimiter

mycli/packages/special/favoritequeries.py

@@ -1,8 +1,8 @@
-# type: ignore
+from __future__ import annotations
 
 
-class FavoriteQueries(object):
-    section_name = "favorite_queries"
+class FavoriteQueries:
+    section_name: str = "favorite_queries"
 
     usage = """
 Favorite Queries are a way to save frequently used queries
@@ -36,27 +36,27 @@ class FavoriteQueries(object):
     # Class-level variable, for convenience to use as a singleton.
     instance = None
 
-    def __init__(self, config):
+    def __init__(self, config) -> None:
         self.config = config
 
     @classmethod
     def from_config(cls, config):
         return FavoriteQueries(config)
 
-    def list(self):
+    def list(self) -> list[str | None]:
         return self.config.get(self.section_name, [])
 
-    def get(self, name):
+    def get(self, name) -> str | None:
         return self.config.get(self.section_name, {}).get(name, None)
 
-    def save(self, name, query):
+    def save(self, name: str, query: str) -> None:
         self.config.encoding = "utf-8"
         if self.section_name not in self.config:
             self.config[self.section_name] = {}
         self.config[self.section_name][name] = query
         self.config.write()
 
-    def delete(self, name):
+    def delete(self, name: str) -> str:
         try:
             del self.config[self.section_name][name]
         except KeyError: