def scan()

in db/postgresql_scanner.py [0:0]

• 37 lines of code
• 8 McCabe index (conditional complexity)

  def scan(self, rds_info, output):
    """Connects to PostgreSql database and collects data.

    Args:
        rds_info: Dictionary object with database connection information
        output: Dictionary object to store the collected data

    Returns:
        True if collection is successful
        False otherwise
    """

    queries = self.get_queries()
    collection = {}
    try:
      conn = psycopg2.connect(
          host=rds_info.host,
          port=rds_info.port,
          database=rds_info.dbname,
          user=rds_info.username,
          password=rds_info.password)

      for query in queries:
        try:
          cur = conn.cursor()
          cur.execute(query.query)
          row_headers = [x[0] for x in cur.description]
          query_results = cur.fetchall()

          if query.query_type == "PostgreSQL_Version":
            version = query_results[0][0]
            collection["version"] = version
            output["version"] = version
          else:
            result_array = []
            for result in query_results:
              result_array.append(
                  dict(zip(row_headers, self.check_result(result))))

            collection[query.query_type] = result_array
        except Exception as ex:   # pylint: disable=broad-except
          if query.query_type == "PostgreSQL_Version":
            raise ex
          logging.error("Failed to run %s", query.query_type)
          logging.error(ex)

      output["PostgreSQL"] = collection
      return True

    except Exception as e:   # pylint: disable=broad-except
      logging.error("Received an unexpected error")
      logging.error(e)
      return False