{"record":{"id":"10710e9bf4a171ad","repo":"redis/redis-py","slug":"get-credentials-must-be-implemented","errorCode":null,"errorMessage":"get_credentials must be implemented","messagePattern":"get_credentials must be implemented","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"redis/credentials.py","lineNumber":14,"sourceCode":"import logging\nfrom abc import ABC, abstractmethod\nfrom typing import Any, Callable, Optional, Tuple, Union\n\nlogger = logging.getLogger(__name__)\n\n\nclass CredentialProvider:\n    \"\"\"\n    Credentials Provider.\n    \"\"\"\n\n    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:\n        raise NotImplementedError(\"get_credentials must be implemented\")\n\n    async def get_credentials_async(self) -> Union[Tuple[str], Tuple[str, str]]:\n        logger.warning(\n            \"This method is added for backward compatibility. \"\n            \"Please override it in your implementation.\"\n        )\n        return self.get_credentials()\n\n\nclass StreamingCredentialProvider(CredentialProvider, ABC):\n    \"\"\"\n    Credential provider that streams credentials in the background.\n    \"\"\"\n\n    @abstractmethod\n    def on_next(self, callback: Callable[[Any], None]):\n        \"\"\"\n        Specifies the callback that should be invoked","sourceCodeStart":1,"sourceCodeEnd":32,"githubUrl":"https://github.com/redis/redis-py/blob/6a6b581b48225afa0b76912d1028c6035baee932/redis/credentials.py#L1-L32","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Override get_credentials(self) in your CredentialProvider subclass to return a (password,) or (username, password) tuple.","If your provider is async-only, also override get_credentials_async so it does not fall back to the unimplemented sync stub.","Prefer reusing UsernamePasswordCredentialProvider for static credentials rather than subclassing the base yourself.","Unit-test the provider in isolation (assert it returns a tuple of the right arity) before wiring it into the client."],"exampleFix":"// before\nclass MyProvider(redis.credentials.CredentialProvider):\n    pass\nclient = redis.Redis(credential_provider=MyProvider())\n// after\nclass MyProvider(redis.credentials.CredentialProvider):\n    def get_credentials(self):\n        tok = fetch_token()  # your rotation logic\n        return (USER, tok)\n    async def get_credentials_async(self):\n        tok = await fetch_token_async()\n        return (USER, tok)","handlingStrategy":"validation","validationCode":"from redis.credentials import CredentialProvider\ndef is_provider_complete(p: CredentialProvider) -> bool:\n    return type(p).get_credentials is not CredentialProvider.get_credentials\nassert is_provider_complete(MyProvider()), 'get_credentials must be implemented'","typeGuard":"from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider\ndef valid_provider(p) -> bool:\n    return isinstance(p, CredentialProvider) and \\\n        type(p).get_credentials is not CredentialProvider.get_credentials","tryCatchPattern":null,"preventionTips":["Prefer UsernamePasswordCredentialProvider for static credentials instead of subclassing the base.","Add a unit test asserting your provider returns a 1- or 2-tuple before wiring it into the client.","Override both get_credentials and get_credentials_async for async-only providers."],"tags":["auth","credentials","implementation","notimplemented"],"backgroundTag":null,"analyzedSha":"6a6b581b48225afa0b76912d1028c6035baee932","analyzedAt":"2026-08-10T12:52:44.840Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-21T04:17:39.646Z"}