{"record":{"id":"1f54eb43f1eaa145","repo":"redis/redis-py","slug":"maintenance-notifications-are-not-supported-by-thi","errorCode":null,"errorMessage":"Maintenance notifications are not supported by this connection type","messagePattern":"Maintenance notifications are not supported by this connection type","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"redis/connection.py","lineNumber":1984,"sourceCode":"\n    @property\n    def _maint_notifications_connection_handler(\n        self,\n    ) -> Optional[MaintNotificationsConnectionHandler]:\n        if isinstance(self._conn, MaintNotificationsAbstractConnection):\n            return self._conn._maint_notifications_connection_handler\n\n    @_maint_notifications_connection_handler.setter\n    def _maint_notifications_connection_handler(\n        self, value: Optional[MaintNotificationsConnectionHandler]\n    ):\n        self._conn._maint_notifications_connection_handler = value\n\n    def _get_socket(self) -> Optional[socket.socket]:\n        if isinstance(self._conn, MaintNotificationsAbstractConnection):\n            return self._conn._get_socket()\n        else:\n            raise NotImplementedError(\n                \"Maintenance notifications are not supported by this connection type\"\n            )\n\n    def _get_maint_notifications_connection_instance(\n        self, connection\n    ) -> MaintNotificationsAbstractConnection:\n        \"\"\"\n        Validate that connection instance supports maintenance notifications.\n        With this helper method we ensure that we are working\n        with the correct connection type.\n        After twe validate that connection instance supports maintenance notifications\n        we can safely return the connection instance\n        as MaintNotificationsAbstractConnection.\n        \"\"\"\n        if not isinstance(connection, MaintNotificationsAbstractConnection):\n            raise NotImplementedError(\n                \"Maintenance notifications are not supported by this connection type\"\n            )","sourceCodeStart":1966,"sourceCodeEnd":2002,"githubUrl":"https://github.com/redis/redis-py/blob/6a6b581b48225afa0b76912d1028c6035baee932/redis/connection.py#L1966-L2002","documentation":"Raised by CacheProxyConnection._get_socket / _get_maint_notifications_connection_instance (connection.py:1980-2004) when the wrapped _conn is not a MaintNotificationsAbstractConnection. Maintenance-notifications state (the live socket, maintenance_state) is only available on connections that implement that abstract base; a plain/legacy connection wrapped by the cache proxy cannot provide it, so NotImplementedError signals an unsupported combination.","triggerScenarios":"Wrapping a connection type that doesn't support maintenance notifications inside CacheProxyConnection, then calling an API that needs the underlying socket or maintenance state (_get_socket, maintenance_state property, etc.). Also reached via _get_maint_notifications_connection_instance when the inner connection isn't the right subclass.","commonSituations":"Mixing the client-side-caching proxy with a connection class that predates maintenance-notifications support; using a custom Connection subclass that doesn't inherit MaintNotificationsAbstractConnection; enabling maintenance-notifications queries on a non-supporting transport.","solutions":["Use the standard redis.Connection (TCP) which is a MaintNotificationsAbstractConnection when wrapping with the cache proxy.","If you wrote a custom Connection, inherit from MaintNotificationsAbstractConnection and implement the required surface.","Do not call maintenance-state APIs (maintenance_state, _get_socket) when the inner connection doesn't support them.","Avoid combining maintenance notifications with connection types that explicitly don't support them."],"exampleFix":"# before\nconn = MyLegacyConnection(...)\ncp = CacheProxyConnection(conn, cache, lock)\ncp._get_socket()\n# after\nfrom redis.connection import Connection\nconn = Connection(host=h, port=p)\ncp = CacheProxyConnection(conn, cache, lock)","handlingStrategy":"type-guard","validationCode":"from redis.connection import Connection, MaintNotificationsAbstractConnection\n# Only wrap connections that support maintenance notifications with the cache proxy\nif isinstance(conn, MaintNotificationsAbstractConnection):\n    cp = CacheProxyConnection(conn, cache, lock)\nelse:\n    raise TypeError('wrap a MaintNotificationsAbstractConnection (e.g. redis.Connection)')","typeGuard":"from redis.connection import MaintNotificationsAbstractConnection\ndef supports_maint_state(conn) -> bool:\n    return isinstance(conn, MaintNotificationsAbstractConnection)","tryCatchPattern":"try:\n    socket = cp._get_socket()\nexcept NotImplementedError:\n    # inner connection can't provide maintenance state; skip those APIs\n    socket = None","preventionTips":["Wrap only standard TCP Connection instances with the cache proxy when using maintenance features.","In custom Connection subclasses, inherit MaintNotificationsAbstractConnection.","Don't call maintenance-state APIs on connection types that don't support them."],"tags":["maintenance-notifications","client-side-caching","connection-type","not-implemented"],"backgroundTag":null,"analyzedSha":"6a6b581b48225afa0b76912d1028c6035baee932","analyzedAt":"2026-08-10T12:52:44.840Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-21T04:17:39.646Z"}