credentials.yaml

yaml

credentials.yml

modeling.yml

llm_api_key: ${OPENAI_API_KEY}

ValueError: Failed to format pattern '${OPENAI_API_KEY}': no config value found, no default provided

TemplatedConfigLoader

OmegaConfigLoader

os.environ.get()

yaml

credentials.yml

.env

python-dotenv

TokenDataSet

open_api_token

TokenDataSet

TokenDataSet

from typing import Any

from kedro_datasets._io import AbstractDataset

class TokenDataSet(AbstractDataset[None, str]):
    def __init__(self, credentials: str):
        self.credentials = credentials

    def _load(self):
        return self.credentials

    def _save(self):
        raise NotImplementedError("Saving TokenDataSet not allowed.")

    def _describe(self) -> dict[str, Any]:
        return dict(type=self.__class__.__name__)

catalog.yml

token_dataset:
  type: yourpackage.datasets.TokenDataSet
  credentials: open_api_token

open_api_token: my_super_secret_token

catalog.load("token_dataset")
>>> 'my_super_secret_token'

OmegaConfigLoader

from kedro.config import OmegaConfigLoader
from omegaconf import OmegaConf
from typing import Any, Dict


class CustomOmegaConfigLoader(OmegaConfigLoader):
    def __init__(
        self,
        conf_source: str,
        env: str = None,
        runtime_params: Dict[str, Any] = None,
    ):
        super().__init__(
            conf_source=conf_source, env=env, runtime_params=runtime_params
        )

        # Register a customer resolver that adds up numbers.
        self.register_custom_resolver("credentials", lambda creds: self.lookup_creds(creds))

    @staticmethod
    def register_custom_resolver(name, function):
        """
        Helper method that checks if the resolver has already been registered and registers the
        resolver if it's new. The check is needed, because omegaconf will throw an error
        if a resolver with the same name is registered twice.
        Alternatively, you can call `register_new_resolver()` with  `replace=True`.
        """
        if not OmegaConf.has_resolver(name):
            OmegaConf.register_new_resolver(name, function)
    
    def lookup_creds(self, creds):
        return self["credentials"][creds]

open_api_key: "${credentials:open_api_token}"