redis/redis-py · error · NotImplementedError
Maintenance notifications are not supported by this…
Error message
Maintenance notifications are not supported by this connection type
What it means
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.
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.
Example fix
# before conn = MyLegacyConnection(...) cp = CacheProxyConnection(conn, cache, lock) cp._get_socket() # after from redis.connection import Connection conn = Connection(host=h, port=p) cp = CacheProxyConnection(conn, cache, lock)
Defensive patterns
Strategy: type-guard
Validate before calling
from redis.connection import Connection, MaintNotificationsAbstractConnection
# Only wrap connections that support maintenance notifications with the cache proxy
if isinstance(conn, MaintNotificationsAbstractConnection):
cp = CacheProxyConnection(conn, cache, lock)
else:
raise TypeError('wrap a MaintNotificationsAbstractConnection (e.g. redis.Connection)') Type guard
from redis.connection import MaintNotificationsAbstractConnection
def supports_maint_state(conn) -> bool:
return isinstance(conn, MaintNotificationsAbstractConnection) Try / catch
try:
socket = cp._get_socket()
except NotImplementedError:
# inner connection can't provide maintenance state; skip those APIs
socket = None Prevention
- 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.
When it happens
Trigger: 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.
Common situations: 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.
Related errors
- Cache must implement CacheInterface
- Cannot create cache key.
- Cannot disable maintenance notifications after enabling them
- Cannot disable maintenance notifications after enabling them
- Cannot enable maintenance notifications for connection…
AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10).
Data as JSON: /api/errors/1f54eb43f1eaa145.
Report an issue: GitHub.
Appendix: source
Thrown at redis/connection.py:1984
@property
def _maint_notifications_connection_handler(
self,
) -> Optional[MaintNotificationsConnectionHandler]:
if isinstance(self._conn, MaintNotificationsAbstractConnection):
return self._conn._maint_notifications_connection_handler
@_maint_notifications_connection_handler.setter
def _maint_notifications_connection_handler(
self, value: Optional[MaintNotificationsConnectionHandler]
):
self._conn._maint_notifications_connection_handler = value
def _get_socket(self) -> Optional[socket.socket]:
if isinstance(self._conn, MaintNotificationsAbstractConnection):
return self._conn._get_socket()
else:
raise NotImplementedError(
"Maintenance notifications are not supported by this connection type"
)
def _get_maint_notifications_connection_instance(
self, connection
) -> MaintNotificationsAbstractConnection:
"""
Validate that connection instance supports maintenance notifications.
With this helper method we ensure that we are working
with the correct connection type.
After twe validate that connection instance supports maintenance notifications
we can safely return the connection instance
as MaintNotificationsAbstractConnection.
"""
if not isinstance(connection, MaintNotificationsAbstractConnection):
raise NotImplementedError(
"Maintenance notifications are not supported by this connection type"
)View on GitHub (pinned to 6a6b581b48)