refctor(mypy): correct type errors

Author: Brooke-white

setup.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 import os
 import sys
+import typing
 
 from setuptools import find_packages, setup
 from setuptools.command.install import install as InstallCommandBase
@@ -10,8 +11,8 @@
 
 
 class BasePytestCommand(TestCommand):
-    user_options = []
-    test_dir = None
+    user_options: typing.List = []
+    test_dir: typing.Optional[str] = None
 
     def initialize_options(self):
         TestCommand.initialize_options(self)
@@ -34,7 +35,7 @@ def run_tests(self):
 
 
 class UnitTestCommand(BasePytestCommand):
-    test_dir = "test/unit"
+    test_dir: str = "test/unit"
 
 
 class IntegrationTestCommand(BasePytestCommand):
@@ -89,7 +90,7 @@ def get_tag(self):
 
 setup(
     name="redshift_connector",
-    version=__version__,
+    version=__version__,  # type: ignore
     description="Redshift interface library",
     long_description=long_description,
     long_description_content_type="text/x-rst",

test/integration/datatype/_generate_test_datatype_tables.py

@@ -39,7 +39,7 @@ class Datatypes(Enum):
 
     @classmethod
     def list(cls) -> typing.List["Datatypes"]:
-        return list(map(lambda p: p, cls))
+        return list(map(lambda p: p, cls))  # type: ignore
 
 
 FLOAT_DATATYPES: typing.Tuple[Datatypes, ...] = (Datatypes.float4, Datatypes.float8)
@@ -52,7 +52,7 @@ def list(cls) -> typing.List["Datatypes"]:
 # 3) (Optional) the Python value we expect to receive. If this field is missing,
 # we expect to receive the test value back directly.
 
-test_data: typing.Dict[Datatypes, typing.Tuple[typing.Tuple[str, ...], ...]] = {
+test_data: typing.Dict[str, typing.Tuple[typing.Tuple[typing.Any, ...], ...]] = {
     Datatypes.int2.name: (  # smallint
         ("-32768", -32768),  # signed 2 byte int min
         ("-128", -128),

test/integration/metadata/test_list_catalog.py

@@ -76,7 +76,7 @@ class TestListCatalog:
     test_cross_db_schema_pattern: str = "p%"
     test_cross_db_table_pattern: str = "a%"
     test_cross_db_column_pattern: str = "c%"
-    test_table_types: typing.Tuple[str] = ("TABLE", "SHARED TABLE", "VIEW", "EXTERNAL TABLE")
+    test_table_types: typing.Tuple[str, ...] = ("TABLE", "SHARED TABLE", "VIEW", "EXTERNAL TABLE")
 
     test_ext_schema: str = "ext_sales"
     test_ext_schema_table: str = "test"
@@ -163,12 +163,12 @@ def test_get_schemas(mocker, _input, db_kwargs):
             assert len(result[0]) == 2
 
 
-def get_tables_test_data() -> typing.List[typing.Tuple[bool, typing.Dict[str, typing.Optional[str]]]]:
-    result: typing.List[typing.Tuple[bool, typing.Dict[str, typing.Optional[str]]]] = []
+def get_tables_test_data() -> typing.List[typing.Optional[typing.Tuple[bool, typing.Dict[str, typing.Any]]]]:
+    result: typing.List[typing.Optional[typing.Tuple[bool, typing.Dict[str, typing.Any]]]] = []
     for flip in (True, False):
         TestListCatalog.config_class_consts(flip)
-        arg_data: typing.List[typing.Dict[str, typing.Optional[str]]] = [
-            {"catalog": None, "schema_pattern": None, "table_name_pattern": None, "types": []},
+        arg_data: typing.List[typing.Dict[str, typing.Any]] = [
+            {"catalog": None, "schema_pattern": None, "table_name_pattern": None, "types": tuple()},
             {
                 "catalog": TestListCatalog.test_db,
                 "schema_pattern": TestListCatalog.test_schema,
@@ -197,7 +197,7 @@ def get_tables_test_data() -> typing.List[typing.Tuple[bool, typing.Dict[str, ty
                 "catalog": TestListCatalog.test_db,
                 "schema_pattern": TestListCatalog.test_schema,
                 "table_name_pattern": TestListCatalog.test_table,
-                "types": [],
+                "types": tuple(),
             },
             {
                 "catalog": None,
@@ -215,15 +215,20 @@ def get_tables_test_data() -> typing.List[typing.Tuple[bool, typing.Dict[str, ty
                 "catalog": TestListCatalog.test_db,
                 "schema_pattern": TestListCatalog.test_schema,
                 "table_name_pattern": None,
-                "types": [],
+                "types": tuple(),
             },
             {
                 "catalog": None,
                 "schema_pattern": TestListCatalog.test_schema,
                 "table_name_pattern": TestListCatalog.test_table,
-                "types": [],
+                "types": tuple(),
+            },
+            {
+                "catalog": None,
+                "schema_pattern": None,
+                "table_name_pattern": TestListCatalog.test_table,
+                "types": tuple(),
             },
-            {"catalog": None, "schema_pattern": None, "table_name_pattern": TestListCatalog.test_table, "types": []},
             {
                 "catalog": None,
                 "schema_pattern": None,
@@ -252,7 +257,7 @@ def get_tables_test_data() -> typing.List[typing.Tuple[bool, typing.Dict[str, ty
                 "catalog": None,
                 "schema_pattern": TestListCatalog.test_schema_pattern,
                 "table_name_pattern": TestListCatalog.test_table_pattern,
-                "types": [],
+                "types": tuple(),
             },
         ]
         for test_case in arg_data:

test/performance/performance.py

@@ -1,6 +1,7 @@
 import configparser
 import os
 import time
+import typing
 
 import redshift_connector
 
@@ -9,12 +10,12 @@
 conf.read(root_path + "/config.ini")
 
 root_path = os.path.dirname(os.path.abspath(__file__))
-sql = open(root_path + "/test.sql", "r", encoding="utf8")
-sqls = sql.readlines()
-sqls = [sql.replace("\n", "") for sql in sqls]
+sql: typing.TextIO = open(root_path + "/test.sql", "r", encoding="utf8")
+sqls: typing.List[str] = sql.readlines()
+sqls = [_sql.replace("\n", "") for _sql in sqls]
 sql.close()
 
-conn = redshift_connector.connect(
+conn: redshift_connector.Connection = redshift_connector.connect(
     database=conf.get("database", "database"),
     host=conf.get("database", "host"),
     port=conf.getint("database", "port"),
@@ -25,11 +26,11 @@
     iam=False,
 )
 
-cursor = conn.cursor()
-for sql in sqls:
-    cursor.execute(sql)
+cursor: redshift_connector.Cursor = conn.cursor()
+for _sql in sqls:
+    cursor.execute(_sql)
 
-result = cursor.fetchall()
+result: typing.Tuple = cursor.fetchall()
 print("fetch {result} rows".format(result=result))
 
 print("start calculate fetch time")