Implement parsing SQLAlchemy objects (#94)

* Implement parsing SQLAlchemy objects

* Update devtools/prettier.py

Co-authored-by: Alex Hall <[email protected]>

* new MetaClass for checking sqlalchemy and some code improvements

* removed json.dumps from _format_sqlalchemy_class function

* Update devtools/prettier.py

Co-authored-by: Alex Hall <[email protected]>

* Update devtools/prettier.py

Co-authored-by: Samuel Colvin <[email protected]>

* add missing import and apply format

* small optimisation to `_format_fields`

Co-authored-by: Alex Hall <[email protected]>
Co-authored-by: Samuel Colvin <[email protected]>

Author: 3 people

.github/workflows/ci.yml

@@ -68,7 +68,7 @@ jobs:
         env_vars: EXTRAS,PYTHON,OS
 
     - name: uninstall extras
-      run: pip uninstall -y multidict numpy pydantic asyncpg
+      run: pip uninstall -y multidict numpy pydantic asyncpg sqlalchemy
 
     - name: test without extras
       run: |

devtools/prettier.py

@@ -3,12 +3,12 @@
 from collections import OrderedDict
 from collections.abc import Generator
 
-from .utils import DataClassType, LaxMapping, env_true, isatty
+from .utils import DataClassType, LaxMapping, SQLAlchemyClassType, env_true, isatty
 
 __all__ = 'PrettyFormat', 'pformat', 'pprint'
 MYPY = False
 if MYPY:
-    from typing import Any, Iterable, Union
+    from typing import Any, Iterable, Tuple, Union
 
 PARENTHESES_LOOKUP = [
     (list, '[', ']'),
@@ -69,6 +69,7 @@ def __init__(
             # put this last as the check can be slow
             (LaxMapping, self._format_dict),
             (DataClassType, self._format_dataclass),
+            (SQLAlchemyClassType, self._format_sqlalchemy_class),
         ]
 
     def __call__(self, value: 'Any', *, indent: int = 0, indent_first: bool = False, highlight: bool = False):
@@ -169,15 +170,7 @@ def _format_tuples(self, value: tuple, value_repr: str, indent_current: int, ind
         fields = getattr(value, '_fields', None)
         if fields:
             # named tuple
-            self._stream.write(value.__class__.__name__ + '(\n')
-            for field, v in zip(fields, value):
-                self._stream.write(indent_new * self._c)
-                if field:  # field is falsy sometimes for odd things like call_args
-                    self._stream.write(str(field))
-                    self._stream.write('=')
-                self._format(v, indent_new, False)
-                self._stream.write(',\n')
-            self._stream.write(indent_current * self._c + ')')
+            self._format_fields(value, zip(fields, value), indent_current, indent_new)
         else:
             # normal tuples are just like other similar iterables
             return self._format_list_like(value, value_repr, indent_current, indent_new)
@@ -231,13 +224,15 @@ def _format_bytearray(self, value: 'Any', _: str, indent_current: int, indent_ne
     def _format_dataclass(self, value: 'Any', _: str, indent_current: int, indent_new: int):
         from dataclasses import asdict
 
-        before_ = indent_new * self._c
-        self._stream.write(f'{value.__class__.__name__}(\n')
-        for k, v in asdict(value).items():
-            self._stream.write(f'{before_}{k}=')
-            self._format(v, indent_new, False)
-            self._stream.write(',\n')
-        self._stream.write(indent_current * self._c + ')')
+        self._format_fields(value, asdict(value).items(), indent_current, indent_new)
+
+    def _format_sqlalchemy_class(self, value: 'Any', _: str, indent_current: int, indent_new: int):
+        fields = [
+            (field, getattr(value, field))
+            for field in dir(value)
+            if not (field.startswith('_') or field in ['metadata', 'registry'])
+        ]
+        self._format_fields(value, fields, indent_current, indent_new)
 
     def _format_raw(self, _: 'Any', value_repr: str, indent_current: int, indent_new: int):
         lines = value_repr.splitlines(True)
@@ -256,6 +251,18 @@ def _format_raw(self, _: 'Any', value_repr: str, indent_current: int, indent_new
         else:
             self._stream.write(value_repr)
 
+    def _format_fields(
+        self, value: 'Any', fields: 'Iterable[Tuple[str, Any]]', indent_current: int, indent_new: int
+    ) -> None:
+        self._stream.write(f'{value.__class__.__name__}(\n')
+        for field, v in fields:
+            self._stream.write(indent_new * self._c)
+            if field:  # field is falsy sometimes for odd things like call_args
+                self._stream.write(f'{field}=')
+            self._format(v, indent_new, False)
+            self._stream.write(',\n')
+        self._stream.write(indent_current * self._c + ')')
+
 
 pformat = PrettyFormat()
 force_highlight = env_true('PY_DEVTOOLS_HIGHLIGHT', None)

devtools/utils.py

@@ -135,3 +135,17 @@ def __instancecheck__(self, instance: 'Any') -> bool:
 
 class DataClassType(metaclass=MetaDataClassType):
     pass
+
+
+class MetaSQLAlchemyClassType(type):
+    def __instancecheck__(self, instance: 'Any') -> bool:
+        try:
+            from sqlalchemy.ext.declarative import DeclarativeMeta
+        except ImportError:
+            return False
+        else:
+            return isinstance(instance.__class__, DeclarativeMeta)
+
+
+class SQLAlchemyClassType(metaclass=MetaSQLAlchemyClassType):
+    pass

tests/requirements.txt

@@ -9,3 +9,4 @@ pydantic
 asyncpg
 numpy
 multidict
+sqlalchemy

tests/test_prettier.py

@@ -27,6 +27,13 @@
 except ImportError:
     Record = None
 
+try:
+    from sqlalchemy import Column, Integer, String
+    from sqlalchemy.ext.declarative import declarative_base
+    SQLAlchemyBase = declarative_base()
+except ImportError:
+    SQLAlchemyBase = None
+
 
 def test_dict():
     v = pformat({1: 2, 3: 4})
@@ -424,3 +431,26 @@ def test_asyncpg_record():
 
 def test_dict_type():
     assert pformat(type({1: 2})) == "<class 'dict'>"
+
+
+@pytest.mark.skipif(SQLAlchemyBase is None, reason='sqlalchemy not installed')
+def test_sqlalchemy_object():
+    class User(SQLAlchemyBase):
+        __tablename__ = 'users'
+        id = Column(Integer, primary_key=True)
+        name = Column(String)
+        fullname = Column(String)
+        nickname = Column(String)
+    user = User()
+    user.id = 1
+    user.name = "Test"
+    user.fullname = "Test For SQLAlchemy"
+    user.nickname = "test"
+    assert pformat(user) == (
+        "User(\n"
+        "    fullname='Test For SQLAlchemy',\n"
+        "    id=1,\n"
+        "    name='Test',\n"
+        "    nickname='test',\n"
+        ")"
+    )