Remove auto show warnings (PyMySQL#774)

Author: methane

pymysql/cursors.py

@@ -2,7 +2,6 @@
 from __future__ import print_function, absolute_import
 from functools import partial
 import re
-import warnings
 
 from ._compat import range_type, text_type, PY2
 from . import err
@@ -35,8 +34,6 @@ class Cursor(object):
     #: Default value of max_allowed_packet is 1048576.
     max_stmt_length = 1024000
 
-    _defer_warnings = False
-
     def __init__(self, connection):
         self.connection = connection
         self.description = None
@@ -46,7 +43,6 @@ def __init__(self, connection):
         self._executed = None
         self._result = None
         self._rows = None
-        self._warnings_handled = False
 
     def close(self):
         """
@@ -90,9 +86,6 @@ def _nextset(self, unbuffered=False):
         """Get the next query set"""
         conn = self._get_db()
         current_result = self._result
-        # for unbuffered queries warnings are only available once whole result has been read
-        if unbuffered:
-            self._show_warnings()
         if current_result is None or current_result is not conn._result:
             return None
         if not current_result.has_next:
@@ -347,26 +340,6 @@ def _do_get_result(self):
         self.description = result.description
         self.lastrowid = result.insert_id
         self._rows = result.rows
-        self._warnings_handled = False
-
-        if not self._defer_warnings:
-            self._show_warnings()
-
-    def _show_warnings(self):
-        if self._warnings_handled:
-            return
-        self._warnings_handled = True
-        if self._result and (self._result.has_next or not self._result.warning_count):
-            return
-        ws = self._get_db().show_warnings()
-        if ws is None:
-            return
-        for w in ws:
-            msg = w[-1]
-            if PY2:
-                if isinstance(msg, unicode):
-                    msg = msg.encode('utf-8', 'replace')
-            warnings.warn(err.Warning(*w[1:3]), stacklevel=4)
 
     def __iter__(self):
         return iter(self.fetchone, None)
@@ -427,8 +400,6 @@ class SSCursor(Cursor):
     possible to scroll backwards, as only the current row is held in memory.
     """
 
-    _defer_warnings = True
-
     def _conv_row(self, row):
         return row
 
@@ -468,7 +439,6 @@ def fetchone(self):
         self._check_executed()
         row = self.read_next()
         if row is None:
-            self._show_warnings()
             return None
         self.rownumber += 1
         return row
@@ -502,7 +472,6 @@ def fetchmany(self, size=None):
         for i in range_type(size):
             row = self.read_next()
             if row is None:
-                self._show_warnings()
                 break
             rows.append(row)
             self.rownumber += 1

pymysql/tests/test_basic.py

@@ -2,7 +2,6 @@
 import datetime
 import json
 import time
-import warnings
 
 import pytest
 
@@ -378,14 +377,3 @@ def test_issue_288(self):
 age = values(age)"""))
         cursor.execute('commit')
         self._verify_records(data)
-
-    def test_warnings(self):
-        con = self.connect()
-        cur = con.cursor()
-        with warnings.catch_warnings(record=True) as ws:
-            warnings.simplefilter("always")
-            cur.execute("drop table if exists no_exists_table")
-        self.assertEqual(len(ws), 1)
-        self.assertEqual(ws[0].category, pymysql.Warning)
-        if u"no_exists_table" not in str(ws[0].message):
-            self.fail("'no_exists_table' not in %s" % (str(ws[0].message),))

pymysql/tests/test_issues.py

@@ -485,28 +485,3 @@ def test_issue_363(self):
         # don't assert the exact internal binary value, as it could
         # vary across implementations
         self.assertTrue(isinstance(row[0], bytes))
-
-    def test_issue_491(self):
-        """ Test warning propagation """
-        conn = pymysql.connect(charset="utf8", **self.databases[0])
-
-        with warnings.catch_warnings():
-            # Ignore all warnings other than pymysql generated ones
-            warnings.simplefilter("ignore")
-            warnings.simplefilter("error", category=pymysql.Warning)
-
-            # verify for both buffered and unbuffered cursor types
-            for cursor_class in (cursors.Cursor, cursors.SSCursor):
-                c = conn.cursor(cursor_class)
-                try:
-                    c.execute("SELECT CAST('124b' AS SIGNED)")
-                    c.fetchall()
-                except pymysql.Warning as e:
-                    # Warnings should have errorcode and string message, just like exceptions
-                    self.assertEqual(len(e.args), 2)
-                    self.assertEqual(e.args[0], 1292)
-                    self.assertTrue(isinstance(e.args[1], text_type))
-                else:
-                    self.fail("Should raise Warning")
-                finally:
-                    c.close()

pymysql/tests/test_load_local.py

@@ -2,7 +2,6 @@
 from pymysql.tests import base
 
 import os
-import warnings
 
 __all__ = ["TestLoadLocal"]
 
@@ -64,29 +63,6 @@ def test_unbuffered_load_file(self):
             c = conn.cursor()
             c.execute("DROP TABLE test_load_local")
 
-    def test_load_warnings(self):
-        """Test load local infile produces the appropriate warnings"""
-        conn = self.connect()
-        c = conn.cursor()
-        c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
-        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
-                                'data',
-                                'load_local_warn_data.txt')
-        try:
-            with warnings.catch_warnings(record=True) as w:
-                warnings.simplefilter('always')
-                c.execute(
-                    ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
-                     "test_load_local FIELDS TERMINATED BY ','").format(filename)
-                )
-                self.assertEqual(w[0].category, Warning)
-                expected_message = "Incorrect integer value"
-                if expected_message not in str(w[-1].message):
-                    self.fail("%r not in %r" % (expected_message, w[-1].message))
-        finally:
-            c.execute("DROP TABLE test_load_local")
-            c.close()
-
 
 if __name__ == "__main__":
     import unittest