redis/redis-py · error · NotImplementedError

get_credentials must be implemented

Error message

get_credentials must be implemented

What it means

Raised as NotImplementedError by CredentialProvider.get_credentials (redis/credentials.py:14). CredentialProvider is the abstract base for pluggable auth (static username/password, EntraID/JWT, etc.); get_credentials() has no default because every real provider must supply credentials differently. Subclassing it without overriding get_credentials() leaves the auth flow unimplemented.

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 da03cdc7e8)

Solutions

  1. Override get_credentials() in your subclass to return a (password,) or (username, password) tuple.
  2. If you only need static credentials, use the built-in UsernamePasswordCredentialProvider instead of subclassing.
  3. For async providers, also override get_credentials_async() to avoid the deprecation warning and blocking fallback.
  4. Make get_credentials() abstract with @abstractmethod so the mistake is caught at instantiation, not at first connect.

Example fix

// before
class MyProvider(redis.credentials.CredentialProvider):
    def __init__(self, vault):
        self.vault = vault
    # forgot get_credentials

// after
class MyProvider(redis.credentials.CredentialProvider):
    def __init__(self, vault):
        self.vault = vault
    def get_credentials(self):
        token = self.vault.get_token()
        return (token,)
Defensive patterns

Strategy: validation

Validate before calling

from redis.credentials import CredentialProvider
class MyProvider(CredentialProvider):
    def get_credentials(self):
        return ('token',)  # or (username, password)
provider = MyProvider()
assert not isinstance(type(provider).get_credentials, type(CredentialProvider.get_credentials)), \
    'get_credentials not overridden'

Type guard

from redis.credentials import CredentialProvider
from typing import Any, cast
import inspect

def is_credential_provider_complete(p: Any) -> bool:
    if not isinstance(p, CredentialProvider):
        return False
    # ensure get_credentials is overridden, not the base stub
    owner = vars(type(p))
    if 'get_credentials' not in owner:
        # check MRO for a concrete (non-base) override
        for klass in type(p).__mro__[1:-1]:
            if 'get_credentials' in vars(klass) and klass is not CredentialProvider:
                return True
        return False
    return True

Prevention

When it happens

Trigger: Creating a custom CredentialProvider subclass (or instantiating the base class) and passing it to a Redis client's credential_provider argument; the first time the connection authenticates it calls get_credentials() and hits the stub.

Common situations: Writing a custom token/secret-manager provider and forgetting to implement the method; copying a provider example that only overrides the async variant; passing CredentialProvider itself instead of UsernamePasswordCredentialProvider.

Related errors


AI-assisted analysis of redis/redis-py@da03cdc7e8 (2026-08-04). Data as JSON: /data/errors/10710e9bf4a171ad.json. Report an issue: GitHub.