🏷️ add types

Author: techouse

mysql_to_sqlite3/cli.py

@@ -1,5 +1,7 @@
 """The command line interface of MySQLtoSQLite."""
+import os
 import sys
+import typing as t
 
 import click
 from tabulate import tabulate
@@ -115,29 +117,29 @@
 @click.option("--debug", is_flag=True, help="Debug mode. Will throw exceptions.")
 @click.version_option(message=tabulate(info(), headers=["software", "version"], tablefmt="github"))
 def cli(
-    sqlite_file,
-    mysql_user,
-    prompt_mysql_password,
-    mysql_password,
-    mysql_database,
-    mysql_tables,
-    exclude_mysql_tables,
-    limit_rows,
-    collation,
-    prefix_indices,
-    without_foreign_keys,
-    without_data,
-    mysql_host,
-    mysql_port,
-    skip_ssl,
-    chunk,
-    log_file,
-    json_as_text,
-    vacuum,
-    use_buffered_cursors,
-    quiet,
-    debug,
-):
+    sqlite_file: t.Union[str, "os.PathLike[t.Any]"],
+    mysql_user: str,
+    prompt_mysql_password: bool,
+    mysql_password: str,
+    mysql_database: str,
+    mysql_tables: t.Optional[t.Sequence[str]],
+    exclude_mysql_tables: t.Optional[t.Sequence[str]],
+    limit_rows: int,
+    collation: t.Optional[str],
+    prefix_indices: bool,
+    without_foreign_keys: bool,
+    without_data: bool,
+    mysql_host: str,
+    mysql_port: int,
+    skip_ssl: bool,
+    chunk: int,
+    log_file: t.Union[str, "os.PathLike[t.Any]"],
+    json_as_text: bool,
+    vacuum: bool,
+    use_buffered_cursors: bool,
+    quiet: bool,
+    debug: bool,
+) -> None:
     """Transfer MySQL to SQLite using the provided CLI options."""
     try:
         if mysql_tables and exclude_mysql_tables:

mysql_to_sqlite3/click_utils.py

@@ -1,5 +1,7 @@
 """Click utilities."""
 
+import typing as t
+
 import click
 
 
@@ -16,7 +18,7 @@ def __init__(self, *args, **kwargs):
         self._previous_parser_process = None
         self._eat_all_parser = None
 
-    def add_to_parser(self, parser, ctx):
+    def add_to_parser(self, parser, ctx) -> None:
         """Override."""
 
         def parser_process(value, state):
@@ -52,7 +54,7 @@ def parser_process(value, state):
         return retval
 
 
-def prompt_password(ctx, param, use_password):  # pylint: disable=W0613
+def prompt_password(ctx: click.core.Context, param: t.Any, use_password: bool):  # pylint: disable=W0613
     """Prompt for password."""
     if use_password:
         mysql_password = ctx.params.get("mysql_password")
@@ -62,7 +64,7 @@ def prompt_password(ctx, param, use_password):  # pylint: disable=W0613
         return mysql_password
 
 
-def validate_positive_integer(ctx, param, value):  # pylint: disable=W0613
+def validate_positive_integer(ctx: click.core.Context, param: t.Any, value: int):  # pylint: disable=W0613
     """Allow only positive integers and 0."""
     if value < 0:
         raise click.BadParameter("Should be a positive integer or 0.")

mysql_to_sqlite3/debug_info.py

@@ -6,6 +6,7 @@
 import platform
 import sqlite3
 import sys
+import typing as t
 from distutils.spawn import find_executable  # pylint: disable=W0402
 from subprocess import check_output
 
@@ -20,7 +21,7 @@
 from . import __version__ as package_version
 
 
-def _implementation():
+def _implementation() -> str:
     """Return a dict with the Python implementation and version.
 
     Provide both the name and the version of the Python implementation
@@ -31,17 +32,17 @@ def _implementation():
     doesn't work for Jython or IronPython. Future investigation should be done
     to work out the correct shape of the code for those platforms.
     """
-    implementation = platform.python_implementation()
+    implementation: str = platform.python_implementation()
 
     if implementation == "CPython":
         implementation_version = platform.python_version()
     elif implementation == "PyPy":
         implementation_version = "%s.%s.%s" % (
-            sys.pypy_version_info.major,  # noqa: ignore=E1101 pylint: disable=E1101
-            sys.pypy_version_info.minor,  # noqa: ignore=E1101 pylint: disable=E1101
-            sys.pypy_version_info.micro,  # noqa: ignore=E1101 pylint: disable=E1101
+            sys.pypy_version_info.major,  # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
+            sys.pypy_version_info.minor,  # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
+            sys.pypy_version_info.micro,  # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
         )
-        rel = sys.pypy_version_info.releaselevel  # noqa: ignore=E1101 pylint: disable=E1101
+        rel = sys.pypy_version_info.releaselevel  # type: ignore # noqa: ignore=E1101 pylint: disable=E1101
         if rel != "final":
             implementation_version = "".join([implementation_version, rel])
     elif implementation == "Jython":
@@ -56,23 +57,23 @@ def _implementation():
     )
 
 
-def _mysql_version():
+def _mysql_version() -> str:
     if find_executable("mysql"):
         try:
-            mysql_version = check_output(["mysql", "-V"])
+            mysql_version: t.Union[str, bytes] = check_output(["mysql", "-V"])
             try:
-                return mysql_version.decode().strip()
+                return mysql_version.decode().strip()  # type: ignore
             except (UnicodeDecodeError, AttributeError):
-                return mysql_version
+                return str(mysql_version)
         except Exception:  # nosec pylint: disable=W0703
             pass
     return "MySQL client not found on the system"
 
 
-def info():
+def info() -> t.List[t.List[str]]:
     """Generate information for a bug report."""
     try:
-        platform_info = "{system} {release}".format(
+        platform_info: str = "{system} {release}".format(
             system=platform.system(),
             release=platform.release(),
         )
@@ -91,7 +92,7 @@ def info():
         ["mysql-connector-python", mysql.connector.__version__],
         ["python-slugify", slugify.__version__],
         ["pytimeparse2", pytimeparse2.__version__],
-        ["simplejson", simplejson.__version__],
+        ["simplejson", simplejson.__version__],  # type: ignore
         ["tabulate", tabulate.__version__],
         ["tqdm", tqdm.__version__],
     ]

mysql_to_sqlite3/py.typed

@@ -1,35 +1,36 @@
 """SQLite adapters and converters for unsupported data types."""
 
 import sqlite3
+import typing as t
 from datetime import date, timedelta
 from decimal import Decimal
 
 from pytimeparse2 import parse
 
 
-def adapt_decimal(value):
+def adapt_decimal(value: t.Any) -> str:
     """Convert decimal.Decimal to string."""
     return str(value)
 
 
-def convert_decimal(value):
-    """Convert string to decimalDecimal."""
+def convert_decimal(value: t.Any) -> Decimal:
+    """Convert string to decimal.Decimal."""
     return Decimal(value)
 
 
-def adapt_timedelta(value):
+def adapt_timedelta(value: t.Any) -> str:
     """Convert datetime.timedelta to %H:%M:%S string."""
     hours, remainder = divmod(value.total_seconds(), 3600)
     minutes, seconds = divmod(remainder, 60)
     return "{:02}:{:02}:{:02}".format(int(hours), int(minutes), int(seconds))
 
 
-def convert_timedelta(value):
+def convert_timedelta(value: t.Any) -> timedelta:
     """Convert %H:%M:%S string to datetime.timedelta."""
     return timedelta(seconds=parse(value))
 
 
-def encode_data_for_sqlite(value):
+def encode_data_for_sqlite(value: t.Any) -> t.Any:
     """Fix encoding bytes."""
     try:
         return value.decode()
@@ -40,12 +41,12 @@ def encode_data_for_sqlite(value):
 class CollatingSequences:
     """Taken from https://www.sqlite.org/datatype3.html#collating_sequences."""
 
-    BINARY = "BINARY"
-    NOCASE = "NOCASE"
-    RTRIM = "RTRIM"
+    BINARY: str = "BINARY"
+    NOCASE: str = "NOCASE"
+    RTRIM: str = "RTRIM"
 
 
-def convert_date(value):
+def convert_date(value: t.Any) -> date:
     """Handle SQLite date conversion."""
     try:
         return date.fromisoformat(value.decode())