Improve SA error handling (#689)

Author: Dreamsorcerer

aiohttp_admin/backends/sqlalchemy.py

@@ -1,7 +1,8 @@
 import asyncio
 import logging
 import operator
-from typing import Any, Iterator, Type, Union
+import sys
+from typing import Any, Callable, Coroutine, Iterator, Type, TypeVar, Union
 
 import sqlalchemy as sa
 from aiohttp import web
@@ -13,6 +14,14 @@
     AbstractAdminResource, CreateParams, DeleteManyParams, DeleteParams, GetListParams,
     GetManyParams, GetOneParams, Record, UpdateManyParams, UpdateParams)
 
+if sys.version_info >= (3, 10):
+    from typing import ParamSpec
+else:
+    from typing_extensions import ParamSpec
+
+_P = ParamSpec("_P")
+_T = TypeVar("_T")
+
 logger = logging.getLogger(__name__)
 
 FIELD_TYPES = {
@@ -26,6 +35,23 @@
 }
 
 
+def handle_errors(
+    f: Callable[_P, Coroutine[None, None, _T]]
+) -> Callable[_P, Coroutine[None, None, _T]]:
+    async def inner(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+        try:
+            return await f(*args, **kwargs)
+        except sa.exc.IntegrityError as e:
+            raise web.HTTPBadRequest(reason=e.args[0])
+        except sa.exc.NoResultFound:
+            logger.warning("No result found (%s)", args, exc_info=True)
+            raise web.HTTPNotFound()
+        except sa.exc.CompileError as e:
+            logger.warning("CompileError (%s)", args, exc_info=True)
+            raise web.HTTPBadRequest(reason=str(e))
+    return inner
+
+
 def create_filters(columns: sa.ColumnCollection[str, sa.Column[object]],
                    filters: dict[str, object]) -> Iterator[ExpressionElementRole[Any]]:
     return (columns[k].in_(v) if isinstance(v, list)
@@ -95,6 +121,7 @@ def __init__(self, db: AsyncEngine, model_or_table: Union[sa.Table, Type[Declara
 
         super().__init__()
 
+    @handle_errors
     async def get_list(self, params: GetListParams) -> tuple[list[Record], int]:
         per_page = params["pagination"]["perPage"]
         offset = (params["pagination"]["page"] - 1) * per_page
@@ -115,16 +142,14 @@ async def get_list(self, params: GetListParams) -> tuple[list[Record], int]:
 
         return entities, count
 
+    @handle_errors
     async def get_one(self, params: GetOneParams) -> Record:
         async with self._db.connect() as conn:
             stmt = sa.select(self._table).where(self._table.c[self.primary_key] == params["id"])
             result = await conn.execute(stmt)
-            try:
-                return result.one()._asdict()
-            except sa.exc.NoResultFound:
-                logger.warning("No result found (%s)", params["id"], exc_info=True)
-                raise web.HTTPNotFound()
+            return result.one()._asdict()
 
+    @handle_errors
     async def get_many(self, params: GetManyParams) -> list[Record]:
         async with self._db.connect() as conn:
             stmt = sa.select(self._table).where(self._table.c[self.primary_key].in_(params["ids"]))
@@ -134,6 +159,7 @@ async def get_many(self, params: GetManyParams) -> list[Record]:
             return records
         raise web.HTTPNotFound()
 
+    @handle_errors
     async def create(self, params: CreateParams) -> Record:
         async with self._db.begin() as conn:
             stmt = sa.insert(self._table).values(params["data"]).returning(*self._table.c)
@@ -144,44 +170,32 @@ async def create(self, params: CreateParams) -> Record:
                 raise web.HTTPBadRequest(reason="Integrity error (element already exists?)")
             return row.one()._asdict()
 
+    @handle_errors
     async def update(self, params: UpdateParams) -> Record:
         async with self._db.begin() as conn:
             stmt = sa.update(self._table).where(self._table.c[self.primary_key] == params["id"])
             stmt = stmt.values(params["data"]).returning(*self._table.c)
-            try:
-                row = await conn.execute(stmt)
-            except sa.exc.CompileError as e:
-                logger.warning("CompileError (%s)", params["id"], exc_info=True)
-                raise web.HTTPBadRequest(reason=str(e))
-            try:
-                return row.one()._asdict()
-            except sa.exc.NoResultFound:
-                logger.warning("No result found (%s)", params["id"], exc_info=True)
-                raise web.HTTPNotFound()
+            row = await conn.execute(stmt)
+            return row.one()._asdict()
 
+    @handle_errors
     async def update_many(self, params: UpdateManyParams) -> list[Union[str, int]]:
         async with self._db.begin() as conn:
             stmt = sa.update(self._table).where(self._table.c[self.primary_key].in_(params["ids"]))
             stmt = stmt.values(params["data"]).returning(self._table.c[self.primary_key])
-            try:
-                r = await conn.scalars(stmt)
-            except sa.exc.CompileError as e:
-                logger.warning("CompileError (%s)", params["ids"], exc_info=True)
-                raise web.HTTPBadRequest(reason=str(e))
+            r = await conn.scalars(stmt)
             # The security check has already called get_many(), so we can be sure
             # there will be results here.
             return list(r)
 
+    @handle_errors
     async def delete(self, params: DeleteParams) -> Record:
         async with self._db.begin() as conn:
             stmt = sa.delete(self._table).where(self._table.c[self.primary_key] == params["id"])
             row = await conn.execute(stmt.returning(*self._table.c))
-            try:
-                return row.one()._asdict()
-            except sa.exc.NoResultFound:
-                logger.warning("No result found (%s)", params["id"], exc_info=True)
-                raise web.HTTPNotFound()
+            return row.one()._asdict()
 
+    @handle_errors
     async def delete_many(self, params: DeleteManyParams) -> list[Union[str, int]]:
         async with self._db.begin() as conn:
             stmt = sa.delete(self._table).where(self._table.c[self.primary_key].in_(params["ids"]))

requirements.txt

@@ -9,3 +9,4 @@ pytest==7.3.1
 pytest-aiohttp==1.0.4
 pytest-cov==4.0.0
 sqlalchemy==2.0.9
+typing_extensions>=3.10; python_version<"3.10"

setup.py

@@ -41,6 +41,7 @@ def read_version():
       license="Apache 2",
       packages=find_packages(),
       install_requires=("aiohttp>=3.8.2", "aiohttp_security", "aiohttp_session",
-                        "cryptography", "pydantic"),
+                        "cryptography", "pydantic",
+                        'typing_extensions>=3.10; python_version<"3.10"'),
       extras_require={"sa": ["sqlalchemy>=2.0.4,<3"]},
       include_package_data=True)