🏷️ fix types

Author: techouse

pyproject.toml

@@ -121,5 +121,5 @@ warn_return_any = true
 warn_unused_configs = true
 
 [[tool.mypy.overrides]]
-module = "pytimeparse2.*"
+module = "pytimeparse2.*,factory.*,docker.*"
 ignore_missing_imports = true

requirements_dev.txt

@@ -13,6 +13,7 @@ simplejson>=3.19.1
 types-simplejson
 sqlalchemy
 sqlalchemy-utils
+types-sqlalchemy-utils
 tox
 tqdm>=4.65.0
 types-tqdm

tests/conftest.py

@@ -19,7 +19,8 @@
 from docker.errors import NotFound
 from docker.models.containers import Container
 from faker import Faker
-from mysql.connector import MySQLConnection, errorcode
+from mysql.connector import CMySQLConnection, MySQLConnection, errorcode
+from mysql.connector.pooling import PooledMySQLConnection
 from requests import HTTPError
 from sqlalchemy import create_engine
 from sqlalchemy.engine import Engine
@@ -236,7 +237,7 @@ def mysql_credentials(pytestconfig: Config) -> MySQLCredentials:
 @pytest.fixture(scope="session")
 def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) -> t.Iterator[MySQLConnection]:
     container: t.Optional[Container] = None
-    mysql_connection: t.Optional[MySQLConnection] = None
+    mysql_connection: t.Optional[t.Union[PooledMySQLConnection, MySQLConnection, CMySQLConnection]] = None
     mysql_available: bool = False
     mysql_connection_retries: int = 15  # failsafe
 
@@ -310,15 +311,15 @@ def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) ->
         if not mysql_available and mysql_connection_retries <= 0:
             raise ConnectionAbortedError("Maximum MySQL connection retries exhausted! Are you sure MySQL is running?")
 
-    yield
+    yield  # type: ignore[misc]
 
     if use_docker and container is not None:
         container.kill()
 
 
 @pytest.fixture()
 def mysql_database(mysql_instance: t.Generator, mysql_credentials: MySQLCredentials) -> t.Iterator[Engine]:
-    yield
+    yield  # type: ignore[misc]
 
     engine: Engine = create_engine(
         "mysql+pymysql://{user}:{password}@{host}:{port}/{database}".format(

tests/func/sqlite3_to_mysql_test.py

@@ -13,7 +13,8 @@
 from _pytest.capture import CaptureFixture
 from _pytest.logging import LogCaptureFixture
 from faker import Faker
-from mysql.connector import MySQLConnection, errorcode
+from mysql.connector import CMySQLConnection, MySQLConnection, errorcode
+from mysql.connector.pooling import PooledMySQLConnection
 from pytest_mock import MockFixture
 from sqlalchemy import MetaData, Table, create_engine, inspect, select, text
 from sqlalchemy.engine import Connection, CursorResult, Engine, Inspector, Row
@@ -31,21 +32,21 @@ class TestSQLite3toMySQL:
     @pytest.mark.parametrize("quiet", [False, True])
     def test_no_sqlite_file_raises_exception(self, quiet: bool) -> None:
         with pytest.raises(ValueError) as excinfo:
-            SQLite3toMySQL(quiet=quiet)
+            SQLite3toMySQL(quiet=quiet)  # type: ignore
         assert "Please provide an SQLite file" in str(excinfo.value)
 
     @pytest.mark.init
     @pytest.mark.parametrize("quiet", [False, True])
     def test_invalid_sqlite_file_raises_exception(self, faker: Faker, quiet: bool) -> None:
         with pytest.raises((FileNotFoundError, IOError)) as excinfo:
-            SQLite3toMySQL(sqlite_file=faker.file_path(depth=1, extension=".sqlite3"), quiet=quiet)
+            SQLite3toMySQL(sqlite_file=faker.file_path(depth=1, extension=".sqlite3"), quiet=quiet)  # type: ignore
         assert "SQLite file does not exist" in str(excinfo.value)
 
     @pytest.mark.init
     @pytest.mark.parametrize("quiet", [False, True])
     def test_missing_mysql_user_raises_exception(self, sqlite_database: str, quiet: bool) -> None:
         with pytest.raises(ValueError) as excinfo:
-            SQLite3toMySQL(sqlite_file=sqlite_database, quiet=quiet)
+            SQLite3toMySQL(sqlite_file=sqlite_database, quiet=quiet)  # type: ignore
         assert "Please provide a MySQL user" in str(excinfo.value)
 
     @pytest.mark.init
@@ -59,7 +60,7 @@ def test_valid_sqlite_file_and_valid_mysql_credentials(
         quiet: bool,
     ):
         with helpers.not_raises(FileNotFoundError):
-            SQLite3toMySQL(
+            SQLite3toMySQL(  # type: ignore
                 sqlite_file=sqlite_database,
                 mysql_user=mysql_credentials.user,
                 mysql_password=mysql_credentials.password,
@@ -81,7 +82,7 @@ def test_valid_sqlite_file_and_invalid_mysql_credentials_raises_access_denied_ex
         quiet: bool,
     ) -> None:
         with pytest.raises(mysql.connector.Error) as excinfo:
-            SQLite3toMySQL(
+            SQLite3toMySQL(  # type: ignore[call-arg]
                 sqlite_file=sqlite_database,
                 mysql_user=faker.first_name().lower(),
                 mysql_password=faker.password(length=16),
@@ -112,7 +113,7 @@ def test_unspecified_mysql_error(
         )
         caplog.set_level(logging.DEBUG)
         with pytest.raises(mysql.connector.Error) as excinfo:
-            SQLite3toMySQL(
+            SQLite3toMySQL(  # type: ignore[call-arg]
                 sqlite_file=sqlite_database,
                 mysql_user=mysql_credentials.user,
                 mysql_password=mysql_credentials.password,
@@ -163,7 +164,7 @@ def cursor(
         mocker.patch.object(mysql.connector, "connect", return_value=FakeMySQLConnection())
         with pytest.raises(mysql.connector.Error):
             caplog.set_level(logging.DEBUG)
-            SQLite3toMySQL(
+            SQLite3toMySQL(  # type: ignore[call-arg]
                 sqlite_file=sqlite_database,
                 mysql_user=mysql_credentials.user,
                 mysql_password=mysql_credentials.password,
@@ -186,7 +187,7 @@ def test_bad_mysql_connection(
             return_value=FakeConnector(is_connected=lambda: False),
         )
         with pytest.raises((ConnectionError, IOError)) as excinfo:
-            SQLite3toMySQL(
+            SQLite3toMySQL(  # type: ignore[call-arg]
                 sqlite_file=sqlite_database,
                 mysql_user=mysql_credentials.user,
                 mysql_password=mysql_credentials.password,
@@ -214,7 +215,7 @@ def test_log_to_file(
         log_file: LocalPath = tmpdir.join(Path("db.log"))
         with pytest.raises(mysql.connector.Error):
             caplog.set_level(logging.DEBUG)
-            SQLite3toMySQL(
+            SQLite3toMySQL(  # type: ignore[call-arg]
                 sqlite_file=sqlite_database,
                 mysql_user=faker.first_name().lower(),
                 mysql_password=faker.password(length=16),
@@ -276,7 +277,7 @@ def test_transfer_transfers_all_tables_in_sqlite_file(
         mysql_insert_method: str,
         ignore_duplicate_keys: bool,
     ):
-        proc: SQLite3toMySQL = SQLite3toMySQL(
+        proc: SQLite3toMySQL = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -317,7 +318,9 @@ def test_transfer_transfers_all_tables_in_sqlite_file(
         mysql_inspect: Inspector = inspect(mysql_engine)
         mysql_tables: t.List[str] = mysql_inspect.get_table_names()
 
-        mysql_connector_connection: MySQLConnection = mysql.connector.connect(
+        mysql_connector_connection: t.Union[
+            PooledMySQLConnection, MySQLConnection, CMySQLConnection
+        ] = mysql.connector.connect(
             user=mysql_credentials.user,
             password=mysql_credentials.password,
             host=mysql_credentials.host,
@@ -339,7 +342,7 @@ def test_transfer_transfers_all_tables_in_sqlite_file(
         """ Test if all the tables have the same indices """
         index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
         mysql_indices: t.Tuple[ReflectedIndex, ...] = tuple(
-            t.cast(ReflectedIndex, {key: index[key] for key in index_keys})
+            t.cast(ReflectedIndex, {key: index[key] for key in index_keys})  # type: ignore[literal-required]
             for index in (chain.from_iterable(mysql_inspect.get_indexes(table_name) for table_name in mysql_tables))
         )
 
@@ -484,7 +487,7 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_sqlite_fi
         random_sqlite_tables: t.List[str] = sample(sqlite_tables, table_number)
         random_sqlite_tables.sort()
 
-        proc: SQLite3toMySQL = SQLite3toMySQL(
+        proc: SQLite3toMySQL = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             sqlite_tables=random_sqlite_tables,
             mysql_user=mysql_credentials.user,
@@ -531,7 +534,7 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_sqlite_fi
         """ Test if all the tables have the same indices """
         index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
         mysql_indices: t.Tuple[ReflectedIndex, ...] = tuple(
-            t.cast(ReflectedIndex, {key: index[key] for key in index_keys})
+            t.cast(ReflectedIndex, {key: index[key] for key in index_keys})  # type: ignore[literal-required]
             for index in (chain.from_iterable(mysql_inspect.get_indexes(table_name) for table_name in mysql_tables))
         )
 

tests/models.py

@@ -7,6 +7,7 @@
 from sqlalchemy import (
     BLOB,
     CHAR,
+    DECIMAL,
     JSON,
     REAL,
     TIMESTAMP,
@@ -25,19 +26,6 @@
 from sqlalchemy.sql.functions import current_timestamp
 
 
-class SQLiteNumeric(types.TypeDecorator):
-    impl: t.Type[String] = types.String
-
-    def load_dialect_impl(self, dialect: Dialect) -> t.Any:
-        return dialect.type_descriptor(types.VARCHAR(100))
-
-    def process_bind_param(self, value: t.Any, dialect: Dialect) -> str:
-        return str(value)
-
-    def process_result_value(self, value: t.Any, dialect: Dialect) -> Decimal:
-        return Decimal(value)
-
-
 class MyCustomType(types.TypeDecorator):
     impl: t.Type[String] = types.String
 
@@ -118,13 +106,13 @@ class Misc(Base):
     char_field: Mapped[str] = mapped_column(CHAR(255), nullable=True)
     date_field: Mapped[date] = mapped_column(nullable=True)
     date_time_field: Mapped[datetime] = mapped_column(nullable=True)
-    decimal_field: Mapped[Decimal] = mapped_column(SQLiteNumeric(10, 2), nullable=True)
-    float_field: Mapped[Decimal] = mapped_column(SQLiteNumeric(12, 4), default=0)
+    decimal_field: Mapped[Decimal] = mapped_column(DECIMAL(10, 2), nullable=True)
+    float_field: Mapped[Decimal] = mapped_column(DECIMAL(12, 4), default=0)
     integer_field: Mapped[int] = mapped_column(default=0)
     if environ.get("LEGACY_DB", "0") == "0":
-        json_field: Mapped[t.Dict[str, t.Any]] = mapped_column(JSON, nullable=True)
-    numeric_field: Mapped[Decimal] = mapped_column(SQLiteNumeric(12, 4), default=0)
-    real_field: Mapped[float] = mapped_column(REAL(12, 4), default=0)
+        json_field: Mapped[t.Mapping[str, t.Any]] = mapped_column(JSON, nullable=True)
+    numeric_field: Mapped[Decimal] = mapped_column(DECIMAL(12, 4), default=0)
+    real_field: Mapped[float] = mapped_column(REAL(12), default=0)
     small_integer_field: Mapped[int] = mapped_column(SmallInteger, default=0)
     string_field: Mapped[str] = mapped_column(String(255), nullable=True)
     text_field: Mapped[str] = mapped_column(Text, nullable=True)

tests/unit/sqlite3_to_mysql_test.py

@@ -27,7 +27,7 @@ def test_translate_type_from_sqlite_to_mysql_invalid_column_type(
         mocker: MockerFixture,
         quiet: bool,
     ) -> None:
-        proc: SQLite3toMySQL = SQLite3toMySQL(
+        proc: SQLite3toMySQL = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -60,7 +60,7 @@ def test_translate_type_from_sqlite_to_mysql_all_valid_columns(
         mysql_string_type: str,
         mysql_text_type: str,
     ) -> None:
-        proc: SQLite3toMySQL = SQLite3toMySQL(
+        proc: SQLite3toMySQL = SQLite3toMySQL(  # type: ignore
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -136,7 +136,7 @@ def test_create_database_connection_error(
         caplog: LogCaptureFixture,
         quiet: bool,
     ) -> None:
-        proc: SQLite3toMySQL = SQLite3toMySQL(
+        proc: SQLite3toMySQL = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -169,7 +169,7 @@ def test_create_table_cursor_error(
         caplog: LogCaptureFixture,
         quiet: bool,
     ) -> None:
-        proc = SQLite3toMySQL(
+        proc = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -208,7 +208,7 @@ def test_process_cursor_error(
         caplog: LogCaptureFixture,
         quiet: bool,
     ) -> None:
-        proc = SQLite3toMySQL(
+        proc = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -240,7 +240,7 @@ def test_add_indices_error(
         caplog: LogCaptureFixture,
         quiet: bool,
     ) -> None:
-        proc = SQLite3toMySQL(
+        proc = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,
@@ -287,7 +287,7 @@ def test_add_foreign_keys_error(
         caplog: LogCaptureFixture,
         quiet: bool,
     ) -> None:
-        proc = SQLite3toMySQL(
+        proc = SQLite3toMySQL(  # type: ignore[call-arg]
             sqlite_file=sqlite_database,
             mysql_user=mysql_credentials.user,
             mysql_password=mysql_credentials.password,