Add pgcode / sqlstate for asyncpg error message

Added accessors ``.sqlstate`` and synonym ``.pgcode`` to the ``.orig``
attribute of the SQLAlchemy exception class raised by the asyncpg DBAPI
adapter, that is, the intermediary exception object that wraps on top of
that raised by the asyncpg library itself, but below the level of the
SQLAlchemy dialect.

Fixes: #6199
Change-Id: Ie0f1ffaaff47c7a50dd1fbccdbe588cdc5322b70

3 files changed, 36 insertions, 0 deletions

diff --git a/doc/build/changelog/unreleased_14/6199.rst b/doc/build/changelog/unreleased_14/6199.rst
new file mode 100644
index 000000000..0c81637f8
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6199.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: usecase, asyncio, postgresql
+    :tickets: 6199
+
+    Added accessors ``.sqlstate`` and synonym ``.pgcode`` to the ``.orig``
+    attribute of the SQLAlchemy exception class raised by the asyncpg DBAPI
+    adapter, that is, the intermediary exception object that wraps on top of
+    that raised by the asyncpg library itself, but below the level of the
+    SQLAlchemy dialect.
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 8cd5bee41..4a191cd28 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -646,6 +646,9 @@ class AsyncAdapt_asyncpg_connection:
                     translated_error = exception_mapping[super_](
                         "%s: %s" % (type(error), error)
                     )
+                    translated_error.pgcode = (
+                        translated_error.sqlstate
+                    ) = getattr(error, "sqlstate", None)
                     raise translated_error from error
             else:
                 raise error
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index f407f2655..1bd947f6a 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -208,6 +208,30 @@ class DialectTest(fixtures.TestBase):
         eq_(cparams["host"], "hostA:portA,hostB,hostC")
 
 
+class PGCodeTest(fixtures.TestBase):
+    __only_on__ = "postgresql"
+
+    def test_error_code(self, metadata, connection):
+        t = Table("t", metadata, Column("id", Integer, primary_key=True))
+        t.create(connection)
+
+        errmsg = assert_raises(
+            exc.IntegrityError,
+            connection.execute,
+            t.insert(),
+            [{"id": 1}, {"id": 1}],
+        )
+
+        if testing.against("postgresql+pg8000"):
+            # TODO: is there another way we're supposed to see this?
+            eq_(errmsg.orig.args[0]["C"], "23505")
+        else:
+            eq_(errmsg.orig.pgcode, "23505")
+
+        if testing.against("postgresql+asyncpg"):
+            eq_(errmsg.orig.sqlstate, "23505")
+
+
 class ExecuteManyMode(object):
     __only_on__ = "postgresql+psycopg2"
     __backend__ = True