Change the dialect for aiopg for better support.

Author: gvbgduh

databases/backends/aiopg.py

@@ -1,10 +1,12 @@
 import getpass
+import json
 import logging
 import typing
 import uuid
 
 import aiopg
-from sqlalchemy.dialects.postgresql import pypostgresql
+from aiopg.sa.engine import APGCompiler_psycopg2
+from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
 from sqlalchemy.engine.interfaces import Dialect, ExecutionContext
 from sqlalchemy.engine.result import ResultMetaData, RowProxy
 from sqlalchemy.sql import ClauseElement
@@ -26,8 +28,10 @@ def __init__(
         self._pool = None
 
     def _get_dialect(self) -> Dialect:
-        dialect = pypostgresql.dialect(paramstyle="pyformat")
-
+        dialect = PGDialect_psycopg2(
+            json_serializer=json.dumps, json_deserializer=lambda x: x
+        )
+        dialect.statement_compiler = APGCompiler_psycopg2
         dialect.implicit_returning = True
         dialect.supports_native_enum = True
         dialect.supports_smallserial = True  # 9.2+

tests/test_databases.py

@@ -2,7 +2,6 @@
 import datetime
 import decimal
 import functools
-import json
 import os
 
 import pytest
@@ -514,14 +513,10 @@ async def test_json_field(database_url):
     async with Database(database_url) as database:
         async with database.transaction(force_rollback=True):
             # execute()
-            query = session.insert()
             data = {"text": "hello", "boolean": True, "int": 1}
             values = {"data": data}
-
-            if database.url.scheme == "postgresql+aiopg":
-                await database.execute(query, {"data": json.dumps(data)})
-            else:
-                await database.execute(query, values)
+            query = session.insert()
+            await database.execute(query, values)
 
             # fetch_all()
             query = session.select()
@@ -544,6 +539,7 @@ async def test_custom_field(database_url):
             # execute()
             query = custom_date.insert()
             values = {"title": "Hello, world", "published": today}
+
             await database.execute(query, values)
 
             # fetch_all()