refactor: Fix precommit hook code formating specified in .pre-commit-config.yaml

Author: rueiyhuang

redshift_connector/__init__.py

@@ -103,6 +103,7 @@
 
 __author__ = "Mathieu Fenniak"
 
+
 def validate_keepalive_values(idle, interval, count):
  if idle is not None:
  if idle <= 0:
@@ -116,6 +117,7 @@ def validate_keepalive_values(idle, interval, count):
  if count <= 0:
  raise ValueError("tcp_keepalive_count must be positive")
 
+
 def connect(
  user: typing.Optional[str] = None,
  database: typing.Optional[str] = None,
@@ -420,11 +422,7 @@ def connect(
 
  if info.tcp_keepalive:
  try:
- validate_keepalive_values(
- info.tcp_keepalive_idle,
- info.tcp_keepalive_interval,
- info.tcp_keepalive_count
- )
+ validate_keepalive_values(info.tcp_keepalive_idle, info.tcp_keepalive_interval, info.tcp_keepalive_count)
  except ValueError as e:
  raise InterfaceError(str(e))
 

redshift_connector/core.py

@@ -2,7 +2,7 @@
 import os
 import socket
 import typing
-from collections import deque, OrderedDict
+from collections import OrderedDict, deque
 from copy import deepcopy
 from datetime import datetime as Datetime
 from datetime import timedelta as Timedelta
@@ -18,8 +18,8 @@
 from scramp import ScramClient # type: ignore
 
 from redshift_connector.config import (
- DEFAULT_PROTOCOL_VERSION,
  DEFAULT_MAX_PREPARED_STATEMENTS,
+ DEFAULT_PROTOCOL_VERSION,
  ClientProtocolVersion,
  DbApiParamstyle,
  _client_encoding,
@@ -358,6 +358,7 @@ def create_message(code: bytes, data: bytes = b"") -> bytes:
 
 arr_trans: typing.Mapping[int, typing.Optional[str]] = dict(zip(map(ord, "[] 'u"), ["{", "}", None, None, None]))
 
+
 class Connection:
  # DBAPI Extension: supply exceptions as attributes on the connection
  Warning = property(lambda self: self._getError(Warning))
@@ -569,9 +570,7 @@ def get_calling_module() -> str:
  "BrowserIdcAuthPlugin",
  ):
  redshift_native_auth = True
- self.set_idc_plugins_params(
- init_params, credentials_provider, identity_namespace, token_type
- )
+ self.set_idc_plugins_params(init_params, credentials_provider, identity_namespace, token_type)
 
  if redshift_native_auth and provider_name:
  init_params["provider_name"] = provider_name
@@ -688,24 +687,24 @@ def get_calling_module() -> str:
  # Set TCP keepalive parameters if supported by platform and values are defined
  if tcp_keepalive_idle is not None:
  # Mac OS X uses TCP_KEEPALIVE instead of TCP_KEEPIDLE
- if hasattr(socket, 'TCP_KEEPIDLE'):
+ if hasattr(socket, "TCP_KEEPIDLE"):
  self._usock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, tcp_keepalive_idle)
  _logger.debug(f"Set TCP_KEEPIDLE to {tcp_keepalive_idle}")
- elif hasattr(socket, 'TCP_KEEPALIVE'): # macOS/BSD
+ elif hasattr(socket, "TCP_KEEPALIVE"): # macOS/BSD
  self._usock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPALIVE, tcp_keepalive_idle)
  _logger.debug(f"Set TCP_KEEPALIVE to {tcp_keepalive_idle}")
  else:
  _logger.warning("Neither TCP_KEEPIDLE nor TCP_KEEPALIVE supported on this platform")
 
  if tcp_keepalive_interval is not None:
- if hasattr(socket, 'TCP_KEEPINTVL'):
+ if hasattr(socket, "TCP_KEEPINTVL"):
  self._usock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, tcp_keepalive_interval)
  _logger.debug(f"Set TCP_KEEPINTVL to {tcp_keepalive_interval}")
  else:
  _logger.warning("TCP_KEEPINTVL not supported on this platform")
 
  if tcp_keepalive_count is not None:
- if hasattr(socket, 'TCP_KEEPCNT'):
+ if hasattr(socket, "TCP_KEEPCNT"):
  self._usock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, tcp_keepalive_count)
  _logger.debug(f"Set TCP_KEEPCNT to {tcp_keepalive_count}")
  else:
@@ -1686,7 +1685,7 @@ def execute(self: "Connection", cursor: Cursor, operation: str, vals) -> None:
  cache = param_cache[pid] = {
  "statement": {},
  "ps": {},
- "statement_dict": OrderedDict() if self.max_prepared_statements > 0 else None
+ "statement_dict": OrderedDict() if self.max_prepared_statements > 0 else None,
  }
 
  try:
@@ -1709,7 +1708,7 @@ def execute(self: "Connection", cursor: Cursor, operation: str, vals) -> None:
  try:
  ps = cache["ps"][key]
  # If statement exists, move it to end of ordered dict (most recently used)
- if self.max_prepared_statements > 0 and 'statement_dict' in cache and key in cache["statement_dict"]:
+ if self.max_prepared_statements > 0 and "statement_dict" in cache and key in cache["statement_dict"]:
  cache["statement_dict"].move_to_end(key)
  _logger.debug("Using cached prepared statement")
  cursor.ps = ps
@@ -2543,6 +2542,8 @@ def get_statement_name_bin(self, statement_name: str) -> bytes:
 
  def get_max_prepared_statement(self, max_prepared_statements: int) -> int:
  if max_prepared_statements < 0:
- _logger.error("Parameter max_prepared_statements must >= 0. Using default value %d", DEFAULT_MAX_PREPARED_STATEMENTS)
+ _logger.error(
+ "Parameter max_prepared_statements must >= 0. Using default value %d", DEFAULT_MAX_PREPARED_STATEMENTS
+ )
  return DEFAULT_MAX_PREPARED_STATEMENTS
  return max_prepared_statements

redshift_connector/cursor.py

@@ -4,12 +4,9 @@
 import typing
 from collections import deque
 from itertools import count, islice
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 from warnings import warn
 
-from redshift_connector.metadataServerAPIHelper import MetadataServerAPIHelper
-from redshift_connector.metadataAPIPostProcessing import MetadataAPIPostProcessing
-
 import redshift_connector
 from redshift_connector.config import (
  ClientProtocolVersion,
@@ -22,6 +19,8 @@
  InterfaceError,
  ProgrammingError,
 )
+from redshift_connector.metadataAPIPostProcessing import MetadataAPIPostProcessing
+from redshift_connector.metadataServerAPIHelper import MetadataServerAPIHelper
 
 if TYPE_CHECKING:
  from redshift_connector.core import Connection
@@ -111,11 +110,11 @@ def __init__(self: "Cursor", connection: "Connection", paramstyle=None) -> None:
 
  self._metadataServerAPIHelper = MetadataServerAPIHelper(self)
  self._metadataAPIPostProcessing = MetadataAPIPostProcessing(self)
- self._cur_catalog: str = None
- self._SHOW_DATABASES_Col_index: typing.Optional = None
- self._SHOW_SCHEMAS_Col_index: typing.Optional = None
- self._SHOW_TABLES_Col_index: typing.Optional = None
- self._SHOW_COLUMNS_Col_index: typing.Optional = None
+ self._cur_catalog: Optional[str] = None
+ self._SHOW_DATABASES_Col_index: Optional[typing.Dict] = None
+ self._SHOW_SCHEMAS_Col_index: Optional[typing.Dict] = None
+ self._SHOW_TABLES_Col_index: Optional[typing.Dict] = None
+ self._SHOW_COLUMNS_Col_index: Optional[typing.Dict] = None
 
  # The minimum show discovery version for the following metadata api was version 2:
  # get_catalogs, get_schemas, get_tables, get_columns
@@ -706,7 +705,9 @@ def _get_catalog_filter_conditions(
 
  catalog_filter: str = ""
  if catalog is not None and catalog != "":
- if self._c.is_single_database_metadata is True or api_supported_only_for_connected_database is True:
+ if (
+ self._c and self._c.is_single_database_metadata is True
+ ) or api_supported_only_for_connected_database is True:
  catalog_filter += " AND current_database() = {catalog}".format(catalog=self.__escape_quotes(catalog))
  else:
  if database_col_name is None or database_col_name == "":
@@ -723,13 +724,19 @@ def get_schemas(
  raise InterfaceError("connection is closed")
 
  if self.supportSHOWDiscovery() >= self._MIN_SHOW_DISCOVERY_VERSION:
- _logger.debug("Support SHOW command. get_schemas with catalog = %s, schema_pattern = %s", catalog, schema_pattern)
+ _logger.debug(
+ "Support SHOW command. get_schemas with catalog = %s, schema_pattern = %s", catalog, schema_pattern
+ )
 
  # Commented out the following line since the Driver will temporarily accept empty string but will block in near future
  # ret_empty: bool = self.is_empty(catalog) or self.is_empty(schema_pattern)
  ret_empty: bool = False
 
- schemas: typing.Tuple = self._metadataAPIPostProcessing.get_schema_post_processing(self._metadataServerAPIHelper.get_schema_server_api(catalog, schema_pattern, ret_empty, self._c.is_single_database_metadata))
+ schemas: typing.Tuple = self._metadataAPIPostProcessing.get_schema_post_processing(
+ self._metadataServerAPIHelper.get_schema_server_api(
+ catalog, schema_pattern, ret_empty, self._c.is_single_database_metadata if self._c else False
+ )
+ )
 
  return schemas
  else:
@@ -740,7 +747,7 @@ def get_schemas_legacy_hardcoded_query(
  ) -> tuple:
  query_args: typing.List[str] = []
  sql: str = ""
- if self._c.is_single_database_metadata is True:
+ if self._c and self._c.is_single_database_metadata is True:
  sql = (
  "SELECT nspname AS TABLE_SCHEM, current_database() AS TABLE_CATALOG FROM pg_catalog.pg_namespace "
  " WHERE nspname <> 'pg_toast' AND (nspname !~ '^pg_temp_' "
@@ -853,22 +860,25 @@ def get_catalogs(self: "Cursor") -> typing.Tuple:
  if self.supportSHOWDiscovery() >= self._MIN_SHOW_DISCOVERY_VERSION:
  _logger.debug("Support SHOW command. get_catalogs")
 
- if self._c.is_single_database_metadata is True:
+ if self._c and self._c.is_single_database_metadata is True:
  sql = "select current_database as TABLE_CAT FROM current_database() ORDER BY TABLE_CAT"
 
  self.execute(sql)
- catalogs: typing.Tuple = self.fetchall()
+ result_catalogs: typing.Tuple = self.fetchall()
  self._metadataAPIPostProcessing.set_row_description(self._metadataAPIPostProcessing._get_catalogs_col)
+ catalogs = result_catalogs
  else:
- catalogs: typing.Tuple = self._metadataAPIPostProcessing.get_catalog_post_processing(self._metadataServerAPIHelper.get_catalog_server_api())
+ catalogs = self._metadataAPIPostProcessing.get_catalog_post_processing(
+ self._metadataServerAPIHelper.get_catalog_server_api()
+ )
 
  return catalogs
  else:
  return self.get_catalogs_legacy_hardcoded_query()
 
  def get_catalogs_legacy_hardcoded_query(self: "Cursor") -> typing.Tuple:
  sql: str = ""
- if self._c.is_single_database_metadata is True:
+ if self._c and self._c.is_single_database_metadata is True:
  sql = "select current_database as TABLE_CAT FROM current_database()"
  else:
  # Datasharing/federation support enable, so get databases using the new view.
@@ -906,13 +916,27 @@ def get_tables(
  types = []
 
  if self.supportSHOWDiscovery() >= self._MIN_SHOW_DISCOVERY_VERSION:
- _logger.debug("Support SHOW command. get_tables with catalog = %s, schema_pattern = %s, table_name_pattern = %s", catalog, schema_pattern, table_name_pattern)
+ _logger.debug(
+ "Support SHOW command. get_tables with catalog = %s, schema_pattern = %s, table_name_pattern = %s",
+ catalog,
+ schema_pattern,
+ table_name_pattern,
+ )
 
  # Commented out the following line since the Driver will temporarily accept empty string but will block in near future
  # ret_empty: bool = self.is_empty(catalog) or self.is_empty(schema_pattern) or self.is_empty(table_name_pattern)
  ret_empty: bool = False
 
- tables: typing.Tuple = self._metadataAPIPostProcessing.get_table_post_processing(self._metadataServerAPIHelper.get_table_server_api(catalog, schema_pattern, table_name_pattern, ret_empty, self._c.is_single_database_metadata), types)
+ tables: typing.Tuple = self._metadataAPIPostProcessing.get_table_post_processing(
+ self._metadataServerAPIHelper.get_table_server_api(
+ catalog,
+ schema_pattern,
+ table_name_pattern,
+ ret_empty,
+ self._c.is_single_database_metadata if self._c else False,
+ ),
+ types,
+ )
 
  return tables
  else:
@@ -931,7 +955,7 @@ def get_tables_legacy_hardcoded_query(
  if schema_pattern_type == "LOCAL_SCHEMA_QUERY":
  sql, sql_args = self.__build_local_schema_tables_query(catalog, schema_pattern, table_name_pattern, types)
  elif schema_pattern_type == "NO_SCHEMA_UNIVERSAL_QUERY":
- if self._c.is_single_database_metadata is True:
+ if self._c and self._c.is_single_database_metadata is True:
  sql, sql_args = self.__build_universal_schema_tables_query(
  catalog, schema_pattern, table_name_pattern, types
  )
@@ -1214,24 +1238,41 @@ def get_columns(
  raise InterfaceError("connection is closed")
 
  if self.supportSHOWDiscovery() >= self._MIN_SHOW_DISCOVERY_VERSION:
- _logger.debug("Support SHOW command. get_columns with catalog = %s, schema_pattern = %s, table_name_pattern = %s, column_name_pattern = %s", catalog, schema_pattern, tablename_pattern, columnname_pattern)
+ _logger.debug(
+ "Support SHOW command. get_columns with catalog = %s, schema_pattern = %s, table_name_pattern = %s, column_name_pattern = %s",
+ catalog,
+ schema_pattern,
+ tablename_pattern,
+ columnname_pattern,
+ )
 
  # Commented out the following line since the Driver will temporarily accept empty string but will block in near future
  # ret_empty: bool = self.is_empty(catalog) or self.is_empty(schema_pattern) or self.is_empty(tablename_pattern) or self.is_empty(columnname_pattern)
  ret_empty: bool = False
 
- columns: typing.Tuple = self._metadataAPIPostProcessing.get_column_post_processing(self._metadataServerAPIHelper.get_column_server_api(catalog, schema_pattern, tablename_pattern, columnname_pattern, ret_empty, self._c.is_single_database_metadata))
+ columns: typing.Tuple = self._metadataAPIPostProcessing.get_column_post_processing(
+ self._metadataServerAPIHelper.get_column_server_api(
+ catalog,
+ schema_pattern,
+ tablename_pattern,
+ columnname_pattern,
+ ret_empty,
+ self._c.is_single_database_metadata if self._c else False,
+ )
+ )
 
  return columns
  else:
- return self.get_columns_legacy_hardcoded_query(catalog, schema_pattern, tablename_pattern, columnname_pattern)
+ return self.get_columns_legacy_hardcoded_query(
+ catalog, schema_pattern, tablename_pattern, columnname_pattern
+ )
 
  def get_columns_legacy_hardcoded_query(
-  self: "Cursor",
-  catalog: typing.Optional[str] = None,
-  schema_pattern: typing.Optional[str] = None,
-  tablename_pattern: typing.Optional[str] = None,
-  columnname_pattern: typing.Optional[str] = None,
+ self: "Cursor",
+ catalog: typing.Optional[str] = None,
+ schema_pattern: typing.Optional[str] = None,
+ tablename_pattern: typing.Optional[str] = None,
+ columnname_pattern: typing.Optional[str] = None,
  ) -> tuple:
  sql: str = ""
  schema_pattern_type: str = self.__schema_pattern_match(schema_pattern)
@@ -1240,7 +1281,7 @@ def get_columns_legacy_hardcoded_query(
  catalog, schema_pattern, tablename_pattern, columnname_pattern
  )
  elif schema_pattern_type == "NO_SCHEMA_UNIVERSAL_QUERY":
- if self._c.is_single_database_metadata is True:
+ if self._c and self._c.is_single_database_metadata is True:
  sql = self.__build_universal_schema_columns_query(
  catalog, schema_pattern, tablename_pattern, columnname_pattern
  )
@@ -2412,7 +2453,7 @@ def __schema_pattern_match(self: "Cursor", schema_pattern: typing.Optional[str])
  if self._c is None:
  raise InterfaceError("connection is closed")
  if schema_pattern is not None and schema_pattern != "":
- if self._c.is_single_database_metadata is True:
+ if self._c and self._c.is_single_database_metadata is True:
  sql: str = "select 1 from svv_external_schemas where schemaname like {schema}".format(
  schema=self.__escape_quotes(schema_pattern)
  )
@@ -2434,6 +2475,8 @@ def __escape_quotes(self: "Cursor", s: str) -> str:
  return "'{s}'".format(s=self.__sanitize_str(s))
 
  def supportSHOWDiscovery(self: "Cursor") -> int:
+ if self._c is None:
+ return 0
  for item in self._c.parameter_statuses:
  if item[0] == b"show_discovery":
  try:
@@ -2462,4 +2505,4 @@ def cur_catalog(self) -> str:
  self._cur_catalog = catalogs[0][0]
  _logger.debug("current catalog: %s", self._cur_catalog)
 
- return self._cur_catalog
+ return self._cur_catalog

redshift_connector/idp_auth_helper.py

@@ -43,7 +43,7 @@ def get_pkg_version(module_name: str) -> Version:
  try:
  from importlib.metadata import version as version
  except ModuleNotFoundError: # if importlib is not present, fallback to pkg_resources
- import pkg_resources
+ import pkg_resources # type: ignore
 
  return Version(pkg_resources.get_distribution(module_name).version)
 
@@ -52,6 +52,7 @@ def get_pkg_version(module_name: str) -> Version:
  return Version(pkg_version)
 
  import importlib
+
  imported_module = importlib.import_module(module_name)
  return Version(imported_module.__version__)