redis/redis-py · error · NotImplementedError

get_credentials must be implemented

Error message

get_credentials must be implemented

What it means

The base class redis.credentials.CredentialProvider leaves get_credentials() unimplemented (redis/credentials.py:14) and raises NotImplementedError to force subclasses to supply credentials. redis-py calls get_credentials() (or the async get_credentials_async()) during the AUTH/HELLO handshake to obtain the (password,) or (username, password) tuple. If you registered a custom credential provider that does not override this method, every connect attempt fails at authentication.

Solutions

  1. Override get_credentials(self) in your CredentialProvider subclass to return a (password,) or (username, password) tuple.
  2. If your provider is async-only, also override get_credentials_async so it does not fall back to the unimplemented sync stub.
  3. Prefer reusing UsernamePasswordCredentialProvider for static credentials rather than subclassing the base yourself.
  4. Unit-test the provider in isolation (assert it returns a tuple of the right arity) before wiring it into the client.

Example fix

// before
class MyProvider(redis.credentials.CredentialProvider):
    pass
client = redis.Redis(credential_provider=MyProvider())
// after
class MyProvider(redis.credentials.CredentialProvider):
    def get_credentials(self):
        tok = fetch_token()  # your rotation logic
        return (USER, tok)
    async def get_credentials_async(self):
        tok = await fetch_token_async()
        return (USER, tok)
Defensive patterns

Strategy: validation

Validate before calling

from redis.credentials import CredentialProvider
def is_provider_complete(p: CredentialProvider) -> bool:
    return type(p).get_credentials is not CredentialProvider.get_credentials
assert is_provider_complete(MyProvider()), 'get_credentials must be implemented'

Type guard

from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
def valid_provider(p) -> bool:
    return isinstance(p, CredentialProvider) and \
        type(p).get_credentials is not CredentialProvider.get_credentials

Prevention

When it happens

Trigger: Instantiating Redis(..., credential_provider=MyProvider()) where MyProvider subclasses CredentialProvider but does not implement get_credentials(); or passing a provider whose get_credentials was renamed/typo'd. The error surfaces on the first command that triggers a connection (connect()/get_connection) because that is when the credentials are read for AUTH.

Common situations: Writing a provider for IAM/EntraID/secret-manager rotation that forgets to implement the synchronous hook; porting from UsernamePasswordCredentialProvider and overriding only __init__; a refactor that renames the method; async-only code paths where the default get_credentials_async falls back to the unimplemented sync method.

Related errors

AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10). Data as JSON: /api/errors/10710e9bf4a171ad. Report an issue: GitHub.

Appendix: source

Thrown at redis/credentials.py:14

import logging
from abc import ABC, abstractmethod
from typing import Any, Callable, Optional, Tuple, Union

logger = logging.getLogger(__name__)


class CredentialProvider:
    """
    Credentials Provider.
    """

    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
        raise NotImplementedError("get_credentials must be implemented")

    async def get_credentials_async(self) -> Union[Tuple[str], Tuple[str, str]]:
        logger.warning(
            "This method is added for backward compatibility. "
            "Please override it in your implementation."
        )
        return self.get_credentials()


class StreamingCredentialProvider(CredentialProvider, ABC):
    """
    Credential provider that streams credentials in the background.
    """

    @abstractmethod
    def on_next(self, callback: Callable[[Any], None]):
        """
        Specifies the callback that should be invoked

View on GitHub (pinned to 6a6b581b48)