Deserialize json type data and Inherit multiple types of cursors at the same time (#343)

* Json Deserialization Cursor
* Inherit multiple types of cursors at the same time
* SS Cursor Modify the result by _conv_row

* Mysql mariadb type is inconsistent
JSON is only an alias for LONGBLOB for compatibility reasons with MySQL
#https://mariadb.com/kb/en/library/json-data-type

* Test against mysql

Author: scp10011

.travis.yml

@@ -26,6 +26,10 @@ matrix:
           env: PYTHONASYNCIODEBUG=
           addons:
               mariadb: 10.1
+        - python: 3.6
+          env: PYTHONASYNCIODEBUG=
+          addons:
+              mysql: 5.7
 
 
 before_script:

aiomysql/connection.py

@@ -37,13 +37,11 @@
 from pymysql.connections import LoadLocalPacketWrapper
 from pymysql.connections import lenenc_int
 
-
 # from aiomysql.utils import _convert_to_str
 from .cursors import Cursor
 from .utils import _ConnectionContextManager, _ContextManager
 from .log import logger
 
-
 DEFAULT_USER = getpass.getuser()
 
 
@@ -389,7 +387,7 @@ def escape_string(self, s):
             return s.replace("'", "''")
         return escape_string(s)
 
-    def cursor(self, cursor=None):
+    def cursor(self, *cursors):
         """Instantiates and returns a cursor
 
         By default, :class:`Cursor` is returned. It is possible to also give a
@@ -402,11 +400,19 @@ def cursor(self, cursor=None):
         """
         self._ensure_alive()
         self._last_usage = self._loop.time()
-        if cursor is not None and not issubclass(cursor, Cursor):
+        try:
+            if cursors and \
+                    any(not issubclass(cursor, Cursor) for cursor in cursors):
+                raise TypeError('Custom cursor must be subclass of Cursor')
+        except TypeError:
             raise TypeError('Custom cursor must be subclass of Cursor')
-
-        if cursor:
-            cur = cursor(self, self._echo)
+        if cursors and len(cursors) == 1:
+            cur = cursors[0](self, self._echo)
+        elif cursors:
+            cursor_name = ''.join(map(lambda x: x.__name__, cursors)) \
+                .replace('Cursor', '') + 'Cursor'
+            cursor_class = type(cursor_name, cursors, {})
+            cur = cursor_class(self, self._echo)
         else:
             cur = self.cursorclass(self, self._echo)
         fut = self._loop.create_future()
@@ -1057,6 +1063,7 @@ def __del__(self):
             warnings.warn("Unclosed connection {!r}".format(self),
                           ResourceWarning)
             self.close()
+
     Warning = Warning
     Error = Error
     InterfaceError = InterfaceError

aiomysql/cursors.py

@@ -1,13 +1,15 @@
 import re
+import json
 import warnings
+import contextlib
 
 from pymysql.err import (
     Warning, Error, InterfaceError, DataError,
     DatabaseError, OperationalError, IntegrityError, InternalError,
     NotSupportedError, ProgrammingError)
 
 from .log import logger
-
+from .connection import FIELD_TYPE
 
 # https://github.com/PyMySQL/PyMySQL/blob/master/pymysql/cursors.py#L11-L18
 
@@ -515,6 +517,41 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         return
 
 
+class _DeserializationCursorMixin:
+    async def _do_get_result(self):
+        await super()._do_get_result()
+        if self._rows:
+            self._rows = [self._deserialization_row(r) for r in self._rows]
+
+    def _deserialization_row(self, row):
+        if row is None:
+            return None
+        if isinstance(row, dict):
+            dict_flag = True
+        else:
+            row = list(row)
+            dict_flag = False
+        for index, (name, field_type, *n) in enumerate(self._description):
+            if field_type == FIELD_TYPE.JSON:
+                point = name if dict_flag else index
+                with contextlib.suppress(ValueError, TypeError):
+                    row[point] = json.loads(row[point])
+        if dict_flag:
+            return row
+        else:
+            return tuple(row)
+
+    def _conv_row(self, row):
+        if row is None:
+            return None
+        row = super()._conv_row(row)
+        return self._deserialization_row(row)
+
+
+class DeserializationCursor(_DeserializationCursorMixin, Cursor):
+    """A cursor automatic deserialization of json type fields"""
+
+
 class _DictCursorMixin:
     # You can override this to use OrderedDict or other dict-like types.
     dict_type = dict
@@ -536,6 +573,7 @@ async def _do_get_result(self):
     def _conv_row(self, row):
         if row is None:
             return None
+        row = super()._conv_row(row)
         return self.dict_type(zip(self._fields, row))
 
 

examples/example_cursors.py

@@ -0,0 +1,20 @@
+import asyncio
+import aiomysql
+
+
+async def test_example(loop):
+    conn = await aiomysql.connect(host='127.0.0.1', port=3306,
+                                  user='root', password='', db='mysql',
+                                  loop=loop)
+    sql = "SELECT 1 `id`, JSON_OBJECT('key1', 1, 'key2', 'abc') obj"
+    async with conn.cursor(aiomysql.cursors.DeserializationCursor,
+                           aiomysql.cursors.DictCursor) as cur:
+        await cur.execute(sql)
+        print(cur.description)
+        r = await cur.fetchall()
+        print(r)
+    conn.close()
+
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(test_example(loop))

tests/test_deserialize_cursor.py

@@ -0,0 +1,175 @@
+import copy
+import asyncio
+
+import aiomysql.cursors
+from tests import base
+from tests._testutils import run_until_complete
+
+
+class TestDeserializeCursor(base.AIOPyMySQLTestCase):
+    bob = ("bob", 21, {"k1": "pretty", "k2": [18, 25]})
+    jim = ("jim", 56, {"k1": "rich", "k2": [20, 60]})
+    fred = ("fred", 100, {"k1": "longevity", "k2": [100, 160]})
+    havejson = True
+
+    cursor_type = aiomysql.cursors.DeserializationCursor
+
+    def setUp(self):
+        super(TestDeserializeCursor, self).setUp()
+        self.conn = conn = self.connections[0]
+
+        @asyncio.coroutine
+        def prepare():
+            c = yield from conn.cursor(self.cursor_type)
+
+            # create a table ane some data to query
+            yield from c.execute("drop table if exists deserialize_cursor")
+            yield from c.execute("select VERSION()")
+            v = yield from c.fetchone()
+            version, *db_type = v[0].split('-', 1)
+            version = float(".".join(version.split('.', 2)[:2]))
+            ismariadb = db_type and 'mariadb' in db_type[0].lower()
+            if ismariadb or version < 5.7:
+                yield from c.execute(
+                    """CREATE TABLE deserialize_cursor
+                     (name char(20), age int , claim text)""")
+                self.havejson = False
+            else:
+                yield from c.execute(
+                    """CREATE TABLE deserialize_cursor
+                     (name char(20), age int , claim json)""")
+            data = [("bob", 21, '{"k1": "pretty", "k2": [18, 25]}'),
+                    ("jim", 56, '{"k1": "rich", "k2": [20, 60]}'),
+                    ("fred", 100, '{"k1": "longevity", "k2": [100, 160]}')]
+            yield from c.executemany("insert into deserialize_cursor values "
+                                     "(%s,%s,%s)",
+                                     data)
+
+        self.loop.run_until_complete(prepare())
+
+    def tearDown(self):
+        @asyncio.coroutine
+        def shutdown():
+            c = yield from self.conn.cursor()
+            yield from c.execute("drop table deserialize_cursor;")
+
+        self.loop.run_until_complete(shutdown())
+        super(TestDeserializeCursor, self).tearDown()
+
+    @run_until_complete
+    def test_deserialize_cursor(self):
+        if not self.havejson:
+            return
+        bob, jim, fred = copy.deepcopy(self.bob), copy.deepcopy(
+            self.jim), copy.deepcopy(self.fred)
+        # all assert test compare to the structure as would come
+        # out from MySQLdb
+        conn = self.conn
+        c = yield from conn.cursor(self.cursor_type)
+
+        # pull back the single row dict for bob and check
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchone()
+        self.assertEqual(bob, r, "fetchone via DeserializeCursor failed")
+        # same again, but via fetchall => tuple)
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchall()
+        self.assertEqual([bob], r,
+                         "fetch a 1 row result via fetchall failed via "
+                         "DeserializeCursor")
+        # get all 3 row via fetchall
+        yield from c.execute("SELECT * from deserialize_cursor")
+        r = yield from c.fetchall()
+        self.assertEqual([bob, jim, fred], r,
+                         "fetchall failed via DictCursor")
+
+        # get all 2 row via fetchmany
+        yield from c.execute("SELECT * from deserialize_cursor")
+        r = yield from c.fetchmany(2)
+        self.assertEqual([bob, jim], r, "fetchmany failed via DictCursor")
+        yield from c.execute('commit')
+
+    @run_until_complete
+    def test_deserialize_cursor_low_version(self):
+        if self.havejson:
+            return
+        bob = ("bob", 21, '{"k1": "pretty", "k2": [18, 25]}')
+        jim = ("jim", 56, '{"k1": "rich", "k2": [20, 60]}')
+        fred = ("fred", 100, '{"k1": "longevity", "k2": [100, 160]}')
+        # all assert test compare to the structure as would come
+        # out from MySQLdb
+        conn = self.conn
+        c = yield from conn.cursor(self.cursor_type)
+
+        # pull back the single row dict for bob and check
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchone()
+        self.assertEqual(bob, r, "fetchone via DeserializeCursor failed")
+        # same again, but via fetchall => tuple)
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchall()
+        self.assertEqual([bob], r,
+                         "fetch a 1 row result via fetchall failed via "
+                         "DeserializeCursor")
+        # get all 3 row via fetchall
+        yield from c.execute("SELECT * from deserialize_cursor")
+        r = yield from c.fetchall()
+        self.assertEqual([bob, jim, fred], r,
+                         "fetchall failed via DictCursor")
+
+        # get all 2 row via fetchmany
+        yield from c.execute("SELECT * from deserialize_cursor")
+        r = yield from c.fetchmany(2)
+        self.assertEqual([bob, jim], r, "fetchmany failed via DictCursor")
+        yield from c.execute('commit')
+
+    @run_until_complete
+    def test_deserializedictcursor(self):
+        if not self.havejson:
+            return
+        bob = {'name': 'bob', 'age': 21,
+               'claim': {"k1": "pretty", "k2": [18, 25]}}
+        conn = self.conn
+        c = yield from conn.cursor(aiomysql.cursors.DeserializationCursor,
+                                   aiomysql.cursors.DictCursor)
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchall()
+        self.assertEqual([bob], r,
+                         "fetch a 1 row result via fetchall failed via "
+                         "DeserializationCursor")
+
+    @run_until_complete
+    def test_ssdeserializecursor(self):
+        if not self.havejson:
+            return
+        conn = self.conn
+        c = yield from conn.cursor(aiomysql.cursors.SSCursor,
+                                   aiomysql.cursors.DeserializationCursor)
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchall()
+        self.assertEqual([self.bob], r,
+                         "fetch a 1 row result via fetchall failed via "
+                         "DeserializationCursor")
+
+    @run_until_complete
+    def test_ssdeserializedictcursor(self):
+        if not self.havejson:
+            return
+        bob = {'name': 'bob', 'age': 21,
+               'claim': {"k1": "pretty", "k2": [18, 25]}}
+        conn = self.conn
+        c = yield from conn.cursor(aiomysql.cursors.SSCursor,
+                                   aiomysql.cursors.DeserializationCursor,
+                                   aiomysql.cursors.DictCursor)
+        yield from c.execute("SELECT * from deserialize_cursor "
+                             "where name='bob'")
+        r = yield from c.fetchall()
+        self.assertEqual([bob], r,
+                         "fetch a 1 row result via fetchall failed via "
+                         "DeserializationCursor")