openstack/deb-python-pymysql
Code Issues Proposed changes

Fix SSDictCursor

Browse Source
This commit is contained in:
INADA Naoki
2013-10-01 03:03:54 +09:00
parent 6c2cf255bc
commit f1ef6e06d8
3 changed files with 39 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
pymysql/cursors.py
View File

@@ -66,6 +66,9 @@ class Cursor(object):
if not self._executed:
raise ProgrammingError("execute() first")
def _conv_row(self, row):
return row
def setinputsizes(self, *args):
"""Does nothing, required by DB API."""
@@ -237,38 +240,24 @@ class DictCursorMixin(object):
# You can override this to use OrderedDict or other dict-like types.
dict_type = dict
def execute(self, query, args=None):
result = super(DictCursorMixin, self).execute(query, args)
def _do_get_result(self):
super(DictCursorMixin, self)._do_get_result()
fields = []
if self.description:
fields = []
for f in self._result.fields:
name = f.name
if name in fields:
name = f.table_name + '.' + name
fields.append(name)
self._fields = fields
return result
def fetchone(self):
''' Fetch the next row '''
result = super(DictCursorMixin, self).fetchone()
if result is None:
return None
return self.dict_type(zip(self._fields, result))
if fields and self._rows:
self._rows = [self._conv_row(r) for r in self._rows]
def fetchmany(self, size=None):
''' Fetch several rows '''
rows = super(DictCursorMixin, self).fetchmany(size)
if rows is None:
def _conv_row(self, row):
if row is None:
return None
return [self.dict_type(zip(self._fields, r)) for r in rows]
def fetchall(self):
''' Fetch all the rows '''
rows = super(DictCursorMixin, self).fetchall()
if rows is None:
return None
return [self.dict_type(zip(self._fields, r)) for r in rows]
return self.dict_type(zip(self._fields, row))
class DictCursor(DictCursorMixin, Cursor):
@@ -291,6 +280,9 @@ class SSCursor(Cursor):
possible to scroll backwards, as only the current row is held in memory.
"""
def _conv_row(self, row):
return row
def close(self):
conn = self.connection
if conn is None:
@@ -317,7 +309,7 @@ class SSCursor(Cursor):
def read_next(self):
""" Read next row """
return self._result._read_rowdata_packet_unbuffered()
return self._conv_row(self._result._read_rowdata_packet_unbuffered())
def fetchone(self):
""" Fetch next row """
@@ -343,7 +335,7 @@ class SSCursor(Cursor):
however, it doesn't make sense to return everything in a list, as that
would use ridiculous memory for large result sets.
"""
return iter(self,fetchone, None)
return iter(self.fetchone, None)
def __iter__(self):
return self.fetchall_unbuffered()

13
pymysql/tests/test_DictCursor.py
View File

@@ -9,12 +9,15 @@ class TestDictCursor(base.PyMySQLTestCase):
jim = {'name': 'jim', 'age': 56, 'DOB': datetime.datetime(1955, 5, 9, 13, 12, 45)}
fred = {'name': 'fred', 'age': 100, 'DOB': datetime.datetime(1911, 9, 12, 1, 1, 1)}
cursor_type = pymysql.cursors.DictCursor
def setUp(self):
super(TestDictCursor, self).setUp()
self.conn = conn = self.connections[0]
c = conn.cursor(pymysql.cursors.DictCursor)
c = conn.cursor(self.cursor_type)
# create a table ane some data to query
c.execute("drop table if exists dictcursor")
c.execute("""CREATE TABLE dictcursor (name char(20), age int , DOB datetime)""")
data = [("bob", 21, "1990-02-06 23:04:56"),
("jim", 56, "1955-05-09 13:12:45"),
@@ -30,7 +33,7 @@ class TestDictCursor(base.PyMySQLTestCase):
bob, jim, fred = self.bob.copy(), self.jim.copy(), self.fred.copy()
#all assert test compare to the structure as would come out from MySQLdb
conn = self.conn
c = conn.cursor(pymysql.cursors.DictCursor)
c = conn.cursor(self.cursor_type)
# try an update which should return no rows
c.execute("update dictcursor set age=20 where name='bob'")
@@ -63,7 +66,7 @@ class TestDictCursor(base.PyMySQLTestCase):
def test_custom_dict(self):
class MyDict(dict): pass
class MyDictCursor(pymysql.cursors.DictCursor):
class MyDictCursor(self.cursor_type):
dict_type = MyDict
keys = ['name', 'age', 'DOB']
@@ -92,6 +95,10 @@ class TestDictCursor(base.PyMySQLTestCase):
"list failed via MyDictCursor")
class TestSSDictCursor(TestDictCursor):
cursor_type = pymysql.cursors.SSDictCursor
if __name__ == "__main__":
import unittest
unittest.main()

13
pymysql/tests/test_issues.py
View File

@@ -19,6 +19,7 @@ class TestOldIssues(base.PyMySQLTestCase):
""" undefined methods datetime_or_None, date_or_None """
conn = self.connections[0]
c = conn.cursor()
c.execute("drop table if exists issue3")
c.execute("create table issue3 (d date, t time, dt datetime, ts timestamp)")
try:
c.execute("insert into issue3 (d, t, dt, ts) values (%s,%s,%s,%s)", (None, None, None, None))
@@ -37,6 +38,7 @@ class TestOldIssues(base.PyMySQLTestCase):
""" can't retrieve TIMESTAMP fields """
conn = self.connections[0]
c = conn.cursor()
c.execute("drop table if exists issue4")
c.execute("create table issue4 (ts timestamp)")
try:
c.execute("insert into issue4 (ts) values (now())")
@@ -65,6 +67,7 @@ class TestOldIssues(base.PyMySQLTestCase):
""" Primary Key and Index error when selecting data """
conn = self.connections[0]
c = conn.cursor()
c.execute("drop table if exists test")
c.execute("""CREATE TABLE `test` (`station` int(10) NOT NULL DEFAULT '0', `dh`
datetime NOT NULL DEFAULT '0000-00-00 00:00:00', `echeance` int(1) NOT NULL
DEFAULT '0', `me` double DEFAULT NULL, `mo` double DEFAULT NULL, PRIMARY
@@ -87,6 +90,7 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
""" can't handle large result fields """
conn = self.connections[0]
cur = conn.cursor()
cur.execute("drop table if exists issue13")
try:
cur.execute("create table issue13 (t text)")
# ticket says 18k
@@ -103,6 +107,7 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
""" query should be expanded before perform character encoding """
conn = self.connections[0]
c = conn.cursor()
c.execute("drop table if exists issue15")
c.execute("create table issue15 (t varchar(32))")
try:
c.execute("insert into issue15 (t) values (%s)", (u'\xe4\xf6\xfc',))
@@ -115,6 +120,7 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
""" Patch for string and tuple escaping """
conn = self.connections[0]
c = conn.cursor()
c.execute("drop table if exists issue16")
c.execute("create table issue16 (name varchar(32) primary key, email varchar(32))")
try:
c.execute("insert into issue16 (name, email) values ('pete', 'floydophone')")
@@ -132,6 +138,7 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
c = conn.cursor()
# grant access to a table to a user with a password
try:
c.execute("drop table if exists issue17")
c.execute("create table issue17 (x varchar(32) primary key)")
c.execute("insert into issue17 (x) values ('hello, world!')")
c.execute("grant all privileges on %s.issue17 to 'issue17user'@'%%' identified by '1234'" % db)
@@ -158,6 +165,7 @@ class TestNewIssues(base.PyMySQLTestCase):
conn = pymysql.connect(charset="utf8", **self.databases[0])
c = conn.cursor()
try:
c.execute(b"drop table if exists hei\xc3\x9fe".decode("utf8"))
c.execute(b"create table hei\xc3\x9fe (name varchar(32))".decode("utf8"))
c.execute(b"insert into hei\xc3\x9fe (name) values ('Pi\xc3\xb1ata')".decode("utf8"))
c.execute(b"select name from hei\xc3\x9fe".decode("utf8"))
@@ -219,6 +227,7 @@ class TestNewIssues(base.PyMySQLTestCase):
datum = "a" * 1024 * 1023 # reduced size for most default mysql installs
try:
c.execute("drop table if exists issue38")
c.execute("create table issue38 (id integer, data mediumblob)")
c.execute("insert into issue38 values (1, %s)", (datum,))
finally:
@@ -227,6 +236,7 @@ class TestNewIssues(base.PyMySQLTestCase):
def disabled_test_issue_54(self):
conn = self.connections[0]
c = conn.cursor()
c.execute("drop table if exists issue54")
big_sql = "select * from issue54 where "
big_sql += " and ".join("%d=%d" % (i,i) for i in range(0, 100000))
@@ -244,6 +254,7 @@ class TestGitHubIssues(base.PyMySQLTestCase):
c = conn.cursor()
self.assertEqual(0, conn.insert_id())
try:
c.execute("drop table if exists issue66")
c.execute("create table issue66 (id integer primary key auto_increment, x integer)")
c.execute("insert into issue66 (x) values (1)")
c.execute("insert into issue66 (x) values (1)")
@@ -280,6 +291,8 @@ class TestGitHubIssues(base.PyMySQLTestCase):
conn = self.connections[0]
c = conn.cursor(pymysql.cursors.DictCursor)
c.execute("drop table if exists a")
c.execute("drop table if exists b")
c.execute("""CREATE TABLE a (id int, value int)""")
c.execute("""CREATE TABLE b (id int, value int)""")