refine codes in runtime (#2764)

sneaxiy · web-flow · commit 67b58337108a · 2020-07-29T16:10:07.000+08:00
diff --git a/python/runtime/db.py b/python/runtime/db.py
@@ -21,8 +21,9 @@
 
 def parseMySQLDSN(dsn):
     # [username[:password]@][protocol[(address)]]/dbname[?param1=value1&...&paramN=valueN]
-    user, passwd, host, port, database, config_str = re.findall(
-        "^(\w*):(\w*)@tcp\(([.a-zA-Z0-9\-]*):([0-9]*)\)/(\w*)(\?.*)?$", dsn)[0]
+    pattern = "^(\w*):(\w*)@tcp\(([.a-zA-Z0-9\-]*):([0-9]*)\)/(\w*)(\?.*)?$"  # noqa: W605, E501
+    found_result = re.findall(pattern, dsn)
+    user, passwd, host, port, database, config_str = found_result[0]
     config = {}
     if len(config_str) > 1:
         for c in config_str[1:].split("&"):
@@ -34,7 +35,7 @@ def parseMySQLDSN(dsn):
 def parseHiveDSN(dsn):
     # usr:pswd@hiveserver:10000/mydb?auth=PLAIN&session.mapreduce_job_queuename=mr
     user_passwd, address_database, config_str = re.findall(
-        "^(.*)@([.a-zA-Z0-9/:_]*)(\?.*)?", dsn)[0]
+        "^(.*)@([.a-zA-Z0-9/:_]*)(\?.*)?", dsn)[0]  # noqa: W605
     user, passwd = user_passwd.split(":")
     if len(address_database.split("/")) > 1:
         address, database = address_database.split("/")
@@ -60,7 +61,7 @@ def parseHiveDSN(dsn):
 def parseMaxComputeDSN(dsn):
     # access_id:access_key@service.com/api?curr_project=test_ci&scheme=http
     user_passwd, address, config_str = re.findall(
-        "^(.*)@([-.a-zA-Z0-9/]*)(\?.*)?", dsn)[0]
+        "^(.*)@([-.a-zA-Z0-9/]*)(\?.*)?", dsn)[0]  # noqa: W605
     user, passwd = user_passwd.split(":")
     config = {}
     if len(config_str) > 1:
@@ -146,6 +147,9 @@ def connect(driver,
     return conn
 
 
+INT64_TYPE = long if six.PY2 else int  # noqa: F821
+
+
 def read_feature(raw_val, feature_spec, feature_name):
     # FIXME(typhoonzero): Should use correct dtype here.
     if feature_spec["is_sparse"]:
@@ -181,7 +185,7 @@ def read_feature(raw_val, feature_spec, feature_name):
     elif feature_spec["dtype"] == "float32":
         return float(raw_val),
     elif feature_spec["dtype"] == "int64":
-        int_raw_val = long(raw_val) if six.PY2 else int(raw_val)
+        int_raw_val = INT64_TYPE(raw_val)
         return int_raw_val,
     elif feature_spec["dtype"] == "string":
         return str(raw_val),
@@ -202,8 +206,8 @@ def limit_select(select, n):
         n (int): the limited row number to query.
 
     Returns:
-        If n >= 0, return a new SQL statement which would query n row(s) at most.
-        If n < 0, return the original SQL statement.
+        If n >= 0, return a new SQL statement which would query n row(s)
+        at most. If n < 0, return the original SQL statement.
     """
     if n < 0:
         return select
@@ -224,7 +228,8 @@ def replace_limit_num(matched_limit):
 
 try:
     import MySQLdb.constants.FIELD_TYPE as MYSQL_FIELD_TYPE
-    # Refer to http://mysql-python.sourceforge.net/MySQLdb-1.2.2/public/MySQLdb.constants.FIELD_TYPE-module.html
+    # Refer to
+    # http://mysql-python.sourceforge.net/MySQLdb-1.2.2/public/MySQLdb.constants.FIELD_TYPE-module.html # noqa: E501
     MYSQL_FIELD_TYPE_DICT = {
         MYSQL_FIELD_TYPE.TINY: "TINYINT",  # 1
         MYSQL_FIELD_TYPE.LONG: "INT",  # 3
@@ -236,7 +241,7 @@ def replace_limit_num(matched_limit):
         MYSQL_FIELD_TYPE.VAR_STRING: "VARCHAR",  # 253
         MYSQL_FIELD_TYPE.STRING: "CHAR",  # 254
     }
-except:
+except:  # noqa: E722
     MYSQL_FIELD_TYPE_DICT = {}
 
 
@@ -369,7 +374,8 @@ def reader():
                 label_idx = reader.field_names.index(
                     label_meta["feature_name"])
             except ValueError:
-                # NOTE(typhoonzero): For clustering model, label_column_name may not in reader.field_names when predicting.
+                # NOTE(typhoonzero): For clustering model, label_column_name
+                # may not in reader.field_names when predicting.
                 label_idx = None
         else:
             label_idx = None
@@ -378,12 +384,14 @@ def reader():
             rows = cursor.fetchmany(size=fetch_size)
             if not rows:
                 break
-            # NOTE: keep the connection while training or connection will lost if no activities appear.
+            # NOTE: keep the connection while training or connection will lost
+            # if no activities appear.
             if driver == "mysql":
                 conn.ping(True)
             for row in rows:
-                # NOTE: If there is no label clause in the extended SQL, the default label value would
-                # be -1, the Model implementation can determine use it or not.
+                # NOTE: If there is no label clause in the extended SQL, the
+                # default label value would be -1, the Model implementation
+                # can determine use it or not.
                 label = row[label_idx] if label_idx is not None else -1
                 if label_meta and label_meta["delimiter"] != "":
                     if label_meta["dtype"] == "float32":
@@ -428,7 +436,7 @@ def reader():
         while True:
             try:
                 row = pai_reader.read(num_records=1)[0]
-            except:
+            except:  # noqa: E722
                 pai_reader.close()
                 break
 
@@ -509,7 +517,7 @@ def execute(conn, sql_stmt):
     Args:
         conn: a database connection, this function will leave it open
         sql_stmt: the sql statement to execute
-    
+
     Returns:
         True on success and False on failure
     """
@@ -522,7 +530,7 @@ def execute(conn, sql_stmt):
             cur.execute(sql_stmt)
             conn.commit()
             return True
-        except:
+        except:  # noqa: E722
             return False
         finally:
             cur.close()
diff --git a/python/runtime/db_test.py b/python/runtime/db_test.py
@@ -17,7 +17,7 @@
 
 import numpy as np
 import runtime.testing as testing
-from odps import ODPS, tunnel
+from odps import tunnel
 from runtime.db import (MYSQL_FIELD_TYPE_DICT, buffered_db_writer, connect,
                         connect_with_data_source, db_generator,
                         get_table_schema, limit_select, parseHiveDSN,
@@ -53,7 +53,7 @@ def execute(driver, conn, statement):
     try:
         rows = cursor.fetchall()
         field_columns = list(map(list, zip(*rows))) if len(rows) > 0 else None
-    except:
+    except:  # noqa: E722
         field_columns = None
 
     return field_names, field_columns
@@ -62,7 +62,9 @@ def execute(driver, conn, statement):
 class TestDB(TestCase):
 
     create_statement = "create table test_db (features text, label int)"
-    hive_create_statement = 'create table test_db (features string, label int) ROW FORMAT DELIMITED FIELDS TERMINATED BY "\001"'
+    hive_create_statement = 'create table test_db (features string, ' \
+                            'label int) ROW FORMAT DELIMITED FIELDS ' \
+                            'TERMINATED BY "\001"'
     select_statement = "select * from test_db"
     drop_statement = "drop table if exists test_db"
 
@@ -114,7 +116,8 @@ def _do_test_hive_specified_db(self,
                                    hdfs_namenode_addr="",
                                    hive_location=""):
         create_db = '''create database if not exists test_db'''
-        create_tbl = '''create table test_db.tbl (features string, label int) ROW FORMAT DELIMITED FIELDS TERMINATED BY "\001"'''
+        create_tbl = '''create table test_db.tbl (features string, label int)
+                        ROW FORMAT DELIMITED FIELDS TERMINATED BY "\001"'''
         drop_tbl = '''drop table if exists test_db.tbl'''
         select_tbl = '''select * from test_db.tbl'''
         table_schema = ["label", "features"]
@@ -173,9 +176,11 @@ def _do_test(self, driver, conn, hdfs_namenode_addr="", hive_location=""):
 
 
 class TestGenerator(TestCase):
-    create_statement = "create table test_table_float_fea (features float, label int)"
+    create_statement = "create table test_table_float_fea " \
+                       "(features float, label int)"
     drop_statement = "drop table if exists test_table_float_fea"
-    insert_statement = "insert into test_table_float_fea (features,label) values(1.0, 0), (2.0, 1)"
+    insert_statement = "insert into test_table_float_fea (features,label)" \
+                       " values(1.0, 0), (2.0, 1)"
 
     @unittest.skipUnless(testing.get_driver() == "mysql",
                          "skip non mysql tests")
@@ -241,29 +246,26 @@ def test_parse_hive_dsn(self):
             ("usr", "pswd", "hiveserver", "1000", "mydb", "PLAIN", {
                 "mapreduce_job_quenename": "mr"
             }),
-            parseHiveDSN(
-                "usr:pswd@hiveserver:1000/mydb?auth=PLAIN&session.mapreduce_job_quenename=mr"
-            ))
+            parseHiveDSN("usr:pswd@hiveserver:1000/mydb?auth=PLAIN&"
+                         "session.mapreduce_job_quenename=mr"))
         self.assertEqual(
             ("usr", "pswd", "hiveserver", "1000", "my_db", "PLAIN", {
                 "mapreduce_job_quenename": "mr"
             }),
-            parseHiveDSN(
-                "usr:pswd@hiveserver:1000/my_db?auth=PLAIN&session.mapreduce_job_quenename=mr"
-            ))
+            parseHiveDSN("usr:pswd@hiveserver:1000/my_db?auth=PLAIN&"
+                         "session.mapreduce_job_quenename=mr"))
         self.assertEqual(
             ("root", "root", "127.0.0.1", None, "mnist", "PLAIN", {}),
             parseHiveDSN("root:root@127.0.0.1/mnist?auth=PLAIN"))
         self.assertEqual(("root", "root", "127.0.0.1", None, None, "", {}),
                          parseHiveDSN("root:root@127.0.0.1"))
 
     def test_parse_maxcompute_dsn(self):
-        self.assertEqual(
-            ("access_id", "access_key", "http://maxcompute-service.com/api",
-             "test_ci"),
-            parseMaxComputeDSN(
-                "access_id:access_key@maxcompute-service.com/api?curr_project=test_ci&scheme=http"
-            ))
+        self.assertEqual(("access_id", "access_key",
+                          "http://maxcompute-service.com/api", "test_ci"),
+                         parseMaxComputeDSN(
+                             "access_id:access_key@maxcompute-service.com/api?"
+                             "curr_project=test_ci&scheme=http"))
 
     def test_kv_feature_column(self):
         feature_spec = {
@@ -300,8 +302,8 @@ def test_get_table_schema(self):
 
             schema = selected_columns_and_types(
                 conn,
-                "SELECT sepal_length, petal_width * 2.3 new_petal_width, class FROM iris.train"
-            )
+                "SELECT sepal_length, petal_width * 2.3 new_petal_width, "
+                "class FROM iris.train")
             expect = [
                 ("sepal_length", "FLOAT"),
                 ("new_petal_width", "DOUBLE"),
@@ -321,8 +323,8 @@ def test_get_table_schema(self):
 
             schema = selected_columns_and_types(
                 conn,
-                "SELECT sepal_length, petal_width * 2.3 AS new_petal_width, class FROM iris.train"
-            )
+                "SELECT sepal_length, petal_width * 2.3 AS new_petal_width, "
+                "class FROM iris.train")
             expect = [
                 ("sepal_length", "FLOAT"),
                 ("new_petal_width", "FLOAT"),
@@ -344,8 +346,8 @@ def test_get_table_schema(self):
 
             schema = selected_columns_and_types(
                 conn,
-                "SELECT sepal_length, petal_width * 2.3 new_petal_width, class FROM %s"
-                % table)
+                "SELECT sepal_length, petal_width * 2.3 new_petal_width, "
+                "class FROM %s" % table)
             expect = [
                 ("sepal_length", "DOUBLE"),
                 ("new_petal_width", "DOUBLE"),
@@ -367,7 +369,8 @@ def test_field_type(self):
 
         table_name = "iris.test_mysql_field_type_table"
         drop_table_sql = "DROP TABLE IF EXISTS %s" % table_name
-        create_table_sql = "CREATE TABLE IF NOT EXISTS " + table_name + "(a %s)"
+        create_table_sql = "CREATE TABLE IF NOT EXISTS " + \
+                           table_name + "(a %s)"
         select_sql = "SELECT * FROM %s" % table_name
 
         for int_type, str_type in MYSQL_FIELD_TYPE_DICT.items():
diff --git a/python/runtime/db_writer/pai_maxcompute.py b/python/runtime/db_writer/pai_maxcompute.py
@@ -11,18 +11,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-try:
-    import paiio
-except:
-    pass
-import tensorflow as tf
-from odps import ODPS, tunnel
-
 from .base import BufferedDBWriter
 
 
 class PAIMaxComputeDBWriter(BufferedDBWriter):
     def __init__(self, table_name, table_schema, buff_size):
+        import paiio
         super(PAIMaxComputeDBWriter, self).__init__(None, table_name,
                                                     table_schema, buff_size)
         table_name_parts = table_name.split(".")
diff --git a/python/runtime/diagnostics.py b/python/runtime/diagnostics.py
@@ -28,7 +28,8 @@ def load_pretrained_model_estimator(estimator,
         estimator_func = estimator.__init__ if inspect.isclass(
             estimator) else estimator
         estimator_spec = inspect.getargspec(estimator_func)
-        # The constructor of Estimator contains named parameter "warm_start_from"
+        # The constructor of Estimator contains named parameter
+        # "warm_start_from"
         warm_start_from_key = "warm_start_from"
         if warm_start_from_key in estimator_spec.args:
             warm_start_from = os.path.abspath(warm_start_from)
diff --git a/python/runtime/explainer.py b/python/runtime/explainer.py
@@ -18,7 +18,8 @@
 import matplotlib.pyplot as plt
 from runtime.oss import copyfileobj
 
-# TODO(shendiaomo): extract common code from tensorflow/explain.py and xgboost/explain.py
+# TODO(shendiaomo): extract common code from tensorflow/explain.py
+# and xgboost/explain.py
 # TODO(shendiaomo): add a unit test for this file later
 
 
diff --git a/python/runtime/maxcompute.py b/python/runtime/maxcompute.py
@@ -65,17 +65,23 @@ def reader():
                     label_idx = reader.field_names.index(
                         label_meta["feature_name"])
                 except ValueError:
-                    # NOTE(typhoonzero): For clustering model, label_column_name may not in reader.field_names when predicting.
+                    # NOTE(typhoonzero): For clustering model,
+                    # label_column_name may not in reader.field_names
+                    # when predicting.
                     label_idx = None
             else:
                 label_idx = None
 
             i = 0
             while i < r.count:
-                expected = r.count - i if r.count - i < fetch_size else fetch_size
+                if r.count - i < fetch_size:
+                    expected = r.count - i
+                else:
+                    expected = fetch_size
                 for row in [[v[1] for v in rec] for rec in r[i:i + expected]]:
-                    # NOTE: If there is no label clause in the extended SQL, the default label value would
-                    # be -1, the Model implementation can determine use it or not.
+                    # NOTE: If there is no label clause in the extended SQL,
+                    # the default label value would be -1, the Model
+                    # implementation can determine use it or not.
                     label = row[label_idx] if label_idx is not None else None
                     if label_meta and label_meta["delimiter"] != "":
                         if label_meta["dtype"] == "float32":
diff --git a/python/runtime/model_metadata.py b/python/runtime/model_metadata.py
@@ -24,7 +24,7 @@ def collect_model_metadata(original_sql, select, validate_select, class_name,
     for (k, v) in attr_copy.items():
         try:
             json.dumps(v)
-        except:
+        except:  # noqa: E722
             attr_copy[k] = str(v)
     metadata['attributes'] = attr_copy
     return metadata
diff --git a/python/runtime/oss.py b/python/runtime/oss.py
