[fea] access to fields via getattr

Author: mosquito

mytor/cursor.py

@@ -103,10 +103,30 @@ def scroll(self, value, mode='relative'):
         return async_call_method(self._cursor.scroll, value, mode)
 
 
-setattr(OriginSSCursor, "__tormysql_class__", SSCursor)
+
+class DBRow(object):
+    __slots__ = ['__row']
+
+    def __init__(self, *args, **kwargs):
+        self.__row = dict(*args, **kwargs)
+
+    def __getattr__(self, item):
+        return self.__row[item]
+
+    def __getitem__(self, item):
+        return self.__row[item]
+
+    def __contains__(self, item):
+        return item in self.__row
+
+    def __repr__(self):
+        return "Row({0})".format(", ".join("{0}={1!r}".format(k, v) for k, v in self.__row.items()))
+
+    def __str__(self):
+        return self.__repr__()
 
 
 class SSDictCursor(SSCursor):
     __delegate_class__ = OriginSSDictCursor
 
-setattr(OriginSSDictCursor, "__tormysql_class__", SSDictCursor)
+DictCursorMixin.dict_type = DBRow

tests/test_with_with.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python
 # encoding: utf-8
+import os
 from tornado.testing import gen_test
+from mytor.cursor import SSDictCursor
 from . import BaseTestCase
 
 
@@ -13,3 +15,28 @@ def test1(self):
                 yield cursor.execute(sql)
                 datas = cursor.fetchall()
                 print (datas)
+
+
+class TestAsyncCursor(BaseTestCase):
+    PARAMS = dict(
+        host=os.getenv("MYSQL_HOST", "127.0.0.1"),
+        user=os.getenv("MYSQL_USER", "root"),
+        passwd=os.getenv("MYSQL_PASSWD", ""),
+        db=os.getenv("MYSQL_DB", "mysql"),
+        charset=os.getenv("MYSQL_CHARSET", "utf8"),
+        no_delay=True,
+        sql_mode="REAL_AS_FLOAT",
+        init_command="SET max_join_size=DEFAULT",
+        cursorclass=SSDictCursor
+    )
+
+    @gen_test
+    def test1(self):
+        sql = "select 1 as test"
+        with (yield self.pool.Connection()) as connection:
+            with connection.cursor() as cursor:
+                yield cursor.execute(sql)
+                result = yield cursor.fetchone()
+                self.assertIn('test', result)
+                self.assertEqual(result['test'], 1)
+                self.assertEqual(result.test, 1)