Refactor Connection.add_notice_callback API.

1. Rename to `Connection.add_log_listener()`
2. Fix `Connection.reset()` to reset log listeners
3. Add a new `PostgresLogMessage` class for log messages
4. Instances of `PostgresLogMessage` are immutable
5. Regenerate `exceptions/__init__`; `PostgresWarning` is now
   a subclass of PostgresLogMessage.

Author: 1st1

asyncpg/connection.py

@@ -42,7 +42,7 @@ class Connection(metaclass=ConnectionMeta):
                  '_stmt_cache', '_stmts_to_close', '_listeners',
                  '_server_version', '_server_caps', '_intro_query',
                  '_reset_query', '_proxy', '_stmt_exclusive_section',
-                 '_config', '_params', '_addr', '_notice_callbacks')
+                 '_config', '_params', '_addr', '_log_listeners')
 
     def __init__(self, protocol, transport, loop,
                  addr: (str, int) or str,
@@ -70,7 +70,7 @@ def __init__(self, protocol, transport, loop,
         self._stmts_to_close = set()
 
         self._listeners = {}
-        self._notice_callbacks = set()
+        self._log_listeners = set()
 
         settings = self._protocol.get_settings()
         ver_string = settings.server_version
@@ -128,25 +128,30 @@ async def remove_listener(self, channel, callback):
             del self._listeners[channel]
             await self.fetch('UNLISTEN {}'.format(channel))
 
-    def add_notice_callback(self, callback):
-        """Add a callback for Postgres notices (NOTICE, DEBUG, LOG etc.).
+    def add_log_listener(self, callback):
+        """Add a listener for Postgres log messages.
 
         It will be called when asyncronous NoticeResponse is received
-        from the connection.  Possible message types are: WARNING, NOTICE, DEBUG,
-        INFO, or LOG.
+        from the connection.  Possible message types are: WARNING, NOTICE,
+        DEBUG, INFO, or LOG.
 
         :param callable callback:
             A callable receiving the following arguments:
             **connection**: a Connection the callback is registered with;
-            **message**: the `exceptions.PostgresNotice` message.
+            **message**: the `exceptions.PostgresLogMessage` message.
+
+        .. versionadded:: 0.12.0
         """
         if self.is_closed():
             raise exceptions.InterfaceError('connection is closed')
-        self._notice_callbacks.add(callback)
+        self._log_listeners.add(callback)
+
+    def remove_log_listener(self, callback):
+        """Remove a listening callback for log messages.
 
-    def remove_notice_callback(self, callback):
-        """Remove a callback for notices."""
-        self._notice_callbacks.discard(callback)
+        .. versionadded:: 0.12.0
+        """
+        self._log_listeners.discard(callback)
 
     def get_server_pid(self):
         """Return the PID of the Postgres server the connection is bound to."""
@@ -975,22 +980,23 @@ async def close(self):
         if self.is_closed():
             return
         self._mark_stmts_as_closed()
-        self._listeners = {}
+        self._listeners.clear()
+        self._log_listeners.clear()
         self._aborted = True
         await self._protocol.close()
-        self._notice_callbacks = set()
 
     def terminate(self):
         """Terminate the connection without waiting for pending data."""
         self._mark_stmts_as_closed()
-        self._listeners = {}
+        self._listeners.clear()
+        self._log_listeners.clear()
         self._aborted = True
         self._protocol.abort()
-        self._notice_callbacks = set()
 
     async def reset(self):
         self._check_open()
         self._listeners.clear()
+        self._log_listeners.clear()
         reset_query = self._get_reset_query()
         if reset_query:
             await self.execute(reset_query)
@@ -1068,44 +1074,37 @@ async def cancel():
 
         self._loop.create_task(cancel())
 
-    def _notice(self, message):
-        if self._proxy is None:
-            con_ref = self
-        else:
-            # See the comment in the `_notify` below.
-            con_ref = self._proxy
+    def _process_log_message(self, fields, last_query):
+        if not self._log_listeners:
+            return
 
-        for cb in self._notice_callbacks:
-            self._loop.call_soon(self._call_notice_cb, cb, con_ref, message)
+        message = exceptions.PostgresLogMessage.new(fields, query=last_query)
 
-    def _call_notice_cb(self, cb, con_ref, message):
+        con_ref = self._unwrap()
+        for cb in self._log_listeners:
+            self._loop.call_soon(
+                self._call_log_listener, cb, con_ref, message)
+
+    def _call_log_listener(self, cb, con_ref, message):
         try:
             cb(con_ref, message)
         except Exception as ex:
             self._loop.call_exception_handler({
-                'message': 'Unhandled exception in asyncpg notice message '
-                           'callback {!r}'.format(cb),
+                'message': 'Unhandled exception in asyncpg log message '
+                           'listener callback {!r}'.format(cb),
                 'exception': ex
             })
 
-    def _notify(self, pid, channel, payload):
+    def _process_notification(self, pid, channel, payload):
         if channel not in self._listeners:
             return
 
-        if self._proxy is None:
-            con_ref = self
-        else:
-            # `_proxy` is not None when the connection is a member
-            # of a connection pool.  Which means that the user is working
-            # with a `PoolConnectionProxy` instance, and expects to see it
-            # (and not the actual Connection) in their event callbacks.
-            con_ref = self._proxy
-
+        con_ref = self._unwrap()
         for cb in self._listeners[channel]:
             self._loop.call_soon(
-                self._call_notify_cb, cb, con_ref, pid, channel, payload)
+                self._call_listener, cb, con_ref, pid, channel, payload)
 
-    def _call_notify_cb(self, cb, con_ref, pid, channel, payload):
+    def _call_listener(self, cb, con_ref, pid, channel, payload):
         try:
             cb(con_ref, pid, channel, payload)
         except Exception as ex:
@@ -1115,6 +1114,17 @@ def _call_notify_cb(self, cb, con_ref, pid, channel, payload):
                 'exception': ex
             })
 
+    def _unwrap(self):
+        if self._proxy is None:
+            con_ref = self
+        else:
+            # `_proxy` is not None when the connection is a member
+            # of a connection pool.  Which means that the user is working
+            # with a `PoolConnectionProxy` instance, and expects to see it
+            # (and not the actual Connection) in their event callbacks.
+            con_ref = self._proxy
+        return con_ref
+
     def _get_reset_query(self):
         if self._reset_query is not None:
             return self._reset_query

asyncpg/exceptions/__init__.py

@@ -5,7 +5,7 @@
 from . import _base
 
 
-class PostgresWarning(Warning, _base.PostgresMessage):
+class PostgresWarning(_base.PostgresLogMessage, Warning):
     sqlstate = '01000'
 
 
@@ -261,6 +261,10 @@ class NumericValueOutOfRangeError(DataError):
     sqlstate = '22003'
 
 
+class SequenceGeneratorLimitExceededError(DataError):
+    sqlstate = '2200H'
+
+
 class StringDataLengthMismatchError(DataError):
     sqlstate = '22026'
 
@@ -608,6 +612,10 @@ class WrongObjectTypeError(SyntaxOrAccessError):
     sqlstate = '42809'
 
 
+class GeneratedAlwaysError(SyntaxOrAccessError):
+    sqlstate = '428C9'
+
+
 class UndefinedColumnError(SyntaxOrAccessError):
     sqlstate = '42703'
 
@@ -772,6 +780,10 @@ class LockNotAvailableError(ObjectNotInPrerequisiteStateError):
     sqlstate = '55P03'
 
 
+class UnsafeNewEnumValueUsageError(ObjectNotInPrerequisiteStateError):
+    sqlstate = '55P04'
+
+
 class OperatorInterventionError(_base.PostgresError):
     sqlstate = '57000'
 
@@ -1007,7 +1019,8 @@ class IndexCorruptedError(InternalServerError):
     'FDWUnableToCreateExecutionError', 'FDWUnableToCreateReplyError',
     'FDWUnableToEstablishConnectionError', 'FeatureNotSupportedError',
     'ForeignKeyViolationError', 'FunctionExecutedNoReturnStatementError',
-    'GroupingError', 'HeldCursorRequiresSameIsolationLevelError',
+    'GeneratedAlwaysError', 'GroupingError',
+    'HeldCursorRequiresSameIsolationLevelError',
     'IdleInTransactionSessionTimeoutError', 'ImplicitZeroBitPadding',
     'InFailedSQLTransactionError',
     'InappropriateAccessModeForBranchTransactionError',
@@ -1073,7 +1086,8 @@ class IndexCorruptedError(InternalServerError):
     'ReadingSQLDataNotPermittedError', 'ReservedNameError',
     'RestrictViolationError', 'SQLRoutineError',
     'SQLStatementNotYetCompleteError', 'SavepointError',
-    'SchemaAndDataStatementMixingNotSupportedError', 'SerializationError',
+    'SchemaAndDataStatementMixingNotSupportedError',
+    'SequenceGeneratorLimitExceededError', 'SerializationError',
     'SnapshotTooOldError', 'SrfProtocolViolatedError',
     'StackedDiagnosticsAccessedWithoutActiveHandlerError',
     'StatementCompletionUnknownError', 'StatementTooComplexError',
@@ -1088,8 +1102,8 @@ class IndexCorruptedError(InternalServerError):
     'UndefinedColumnError', 'UndefinedFileError',
     'UndefinedFunctionError', 'UndefinedObjectError',
     'UndefinedParameterError', 'UndefinedTableError',
-    'UniqueViolationError', 'UnterminatedCStringError',
-    'UntranslatableCharacterError', 'WindowingError',
-    'WithCheckOptionViolationError', 'WrongObjectTypeError',
-    'ZeroLengthCharacterStringError'
+    'UniqueViolationError', 'UnsafeNewEnumValueUsageError',
+    'UnterminatedCStringError', 'UntranslatableCharacterError',
+    'WindowingError', 'WithCheckOptionViolationError',
+    'WrongObjectTypeError', 'ZeroLengthCharacterStringError'
 )

asyncpg/exceptions/_base.py

@@ -5,11 +5,12 @@
 # the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0
 
 
+import asyncpg
 import sys
 
 
 __all__ = ('PostgresError', 'FatalPostgresError', 'UnknownPostgresError',
-           'InterfaceError', 'PostgresNotice')
+           'InterfaceError', 'PostgresLogMessage')
 
 
 def _is_asyncpg_class(cls):
@@ -18,6 +19,7 @@ def _is_asyncpg_class(cls):
 
 
 class PostgresMessageMeta(type):
+
     _message_map = {}
     _field_map = {
         'S': 'severity',
@@ -69,35 +71,31 @@ def get_message_class_for_sqlstate(mcls, code):
 
 
 class PostgresMessage(metaclass=PostgresMessageMeta):
-    def __str__(self):
-        msg = self.args[0]
-        if self.detail:
-            msg += '\nDETAIL:  {}'.format(self.detail)
-        if self.hint:
-            msg += '\nHINT:  {}'.format(self.hint)
 
-        return msg
+    @classmethod
+    def _get_error_class(cls, fields):
+        sqlstate = fields.get('C')
+        return type(cls).get_message_class_for_sqlstate(sqlstate)
 
     @classmethod
-    def _get_error_template(cls, fields, query):
-        errcode = fields.get('C')
-        mcls = cls.__class__
-        exccls = mcls.get_message_class_for_sqlstate(errcode)
+    def _get_error_dict(cls, fields, query):
         dct = {
             'query': query
         }
 
+        field_map = type(cls)._field_map
         for k, v in fields.items():
-            field = mcls._field_map.get(k)
+            field = field_map.get(k)
             if field:
                 dct[field] = v
 
-        return exccls, dct
+        return dct
 
     @classmethod
-    def new(cls, fields, query=None):
-        exccls, dct = cls._get_error_template(fields, query)
+    def _make_constructor(cls, fields, query=None):
+        dct = cls._get_error_dict(fields, query)
 
+        exccls = cls._get_error_class(fields)
         message = dct.get('message', '')
 
         # PostgreSQL will raise an exception when it detects
@@ -122,24 +120,36 @@ def new(cls, fields, query=None):
             message = ('cached statement plan is invalid due to a database '
                        'schema or configuration change')
 
-        e = exccls(message)
-        e.__dict__.update(dct)
-
-        return e
+        return exccls, message, dct
 
     def as_dict(self):
-        message = {}
+        dct = {}
         for f in type(self)._field_map.values():
             val = getattr(self, f)
             if val is not None:
-                message[f] = val
-
-        return message
+                dct[f] = val
+        return dct
 
 
 class PostgresError(PostgresMessage, Exception):
     """Base class for all Postgres errors."""
 
+    def __str__(self):
+        msg = self.args[0]
+        if self.detail:
+            msg += '\nDETAIL:  {}'.format(self.detail)
+        if self.hint:
+            msg += '\nHINT:  {}'.format(self.hint)
+
+        return msg
+
+    @classmethod
+    def new(cls, fields, query=None):
+        exccls, message, dct = cls._make_constructor(fields, query)
+        ex = exccls(message)
+        ex.__dict__.update(dct)
+        return ex
+
 
 class FatalPostgresError(PostgresError):
     """A fatal error that should result in server disconnection."""
@@ -153,8 +163,32 @@ class InterfaceError(Exception):
     """An error caused by improper use of asyncpg API."""
 
 
-class PostgresNotice(PostgresMessage):
-    sqlstate = '00000'
+class PostgresLogMessage(PostgresMessage):
+    """A base class for non-error server messages."""
+
+    def __str__(self):
+        return '{}: {}'.format(type(self).__name__, self.message)
 
-    def __init__(self, message):
-        self.args = [message]
+    def __setattr__(self, name, val):
+        raise TypeError('instances of {} are immutable'.format(
+            type(self).__name__))
+
+    @classmethod
+    def new(cls, fields, query=None):
+        exccls, message_text, dct = cls._make_constructor(fields, query)
+
+        if exccls is UnknownPostgresError:
+            exccls = PostgresLogMessage
+
+        if exccls is PostgresLogMessage:
+            severity = dct.get('severity_en') or dct.get('severity')
+            if severity and severity.upper() == 'WARNING':
+                exccls = asyncpg.PostgresWarning
+
+        if issubclass(exccls, (BaseException, Warning)):
+            msg = exccls(message_text)
+        else:
+            msg = exccls()
+
+        msg.__dict__.update(dct)
+        return msg