Make credentials a resolver

[Galileo-Galilei]

Description

Make "credentials" a OmegaConf resolver, so injecting credentials in the catalog would look like :

# current injection
my_data:
    type: pandas.CSVDataset
    filepath: /path/to/data.csv
    credentials: my_credentials # a string which refer to a a section in credentials.yaml

# new proposal
my_data:
    type: pandas.CSVDataset
    filepath: /path/to/data.csv
    credentials: ${credentials: my_credentials}

Context

This may look like a pure technical refactoring, but actually I am expecting this to be an actual improvement for users, e.g. :

• improve readability : credentials are hard to grasp because it does some magic relying on the template. As of our old kedro motto A sprinkle of magic is better than a spoonful of it or the Zen of python Explicit is better than implicit, above syntax will help people to understand what is happening under the hood.
• define logic for empty credentials : users will be able to handle "empty" credentials (often not provided when developing locally but necessary in production, this forces you to have empty keys during development), see Better Credentials Support in Catalog #3634 for a user monkey patching credentials in settings.py
• pass credentials object instead of strings: it is common to have some complicated credentials objects which are hard to express purely with a yaml description, see Adding BigQuery authentication to credentials.yml #1621
• enable custom logic to fetch credentials, which is often hardcoded in hooks right now : Add an advanced example (how to use hooks to load external credentials) to the configuration documentation #2547
• improve sharing logic : a credentials is just a normal python function. Passing it as a resolver would enable people to create their own credentials fetching logic and override kedro's just by installing a package provided by their org.
• decouple the code of catalog and the config resolution: all the logic currently lives in the Datacatalog class (see below) , which does not look right. Moving this logic to the config would eventually helps Users cannot install specific components of Kedro separately #3659 if we want to go forward in this direction.

Overall, this would likely be a general improvement with some overlaps with what is described here #1646 and has never been properly adressed, despite a bunch of discussions around this topic.

Possible Implementation

Move the logic from the catalog and the _get_credentials and _resolve_credentials:

kedro/kedro/io/data_catalog.py

Lines 35 to 58 in 7237dc7

	def _get_credentials(credentials_name: str, credentials: dict[str, Any]) -> Any:
	"""Return a set of credentials from the provided credentials dict.
	Args:
	credentials_name: Credentials name.
	credentials: A dictionary with all credentials.
	Returns:
	The set of requested credentials.
	Raises:
	KeyError: When a data set with the given name has not yet been
	registered.
	"""
	try:
	return credentials[credentials_name]
	except KeyError as exc:
	raise KeyError(
	f"Unable to find credentials '{credentials_name}': check your data "
	"catalog and credentials configuration. See "
	"https://kedro.readthedocs.io/en/stable/kedro.io.DataCatalog.html "
	"for an example."
	) from exc

kedro/kedro/io/data_catalog.py

Lines 61 to 83 in 7237dc7

	def _resolve_credentials(
	config: dict[str, Any], credentials: dict[str, Any]
	) -> dict[str, Any]:
	"""Return the dataset configuration where credentials are resolved using
	credentials dictionary provided.
	Args:
	config: Original dataset config, which may contain unresolved credentials.
	credentials: A dictionary with all credentials.
	Returns:
	The dataset config, where all the credentials are successfully resolved.
	"""
	config = copy.deepcopy(config)
	def _map_value(key: str, value: Any) -> Any:
	if key == CREDENTIALS_KEY and isinstance(value, str):
	return _get_credentials(value, credentials)
	if isinstance(value, dict):
	return {k: _map_value(k, v) for k, v in value.items()}
	return value
	return {k: _map_value(k, v) for k, v in config.items()}

to a resolver.

Possible Alternatives

Do nothing and keep the situation as is.

[idanov]

I like the idea in principle, but what would be a way for us to prevent hardcoding of credentials into the catalog directly?

[Galileo-Galilei]

Not sure of what you mean : for me the default credentials provided by kedro would do the exact same things as it does now (map a string in the catalog to an entry in credentials.yml) so I don't think there is a problem here.

A user who likes playing with kedro's limits could eventually write his own resolver and eventually takes raw inputs directly in the catalog, but I don't think it's kedro's responsibility to try to circumvent all weird things users can do (and funnily, they already can write a custom resolver and use it in the catalog - my proposal would simply enable to make their dangerous logic the default, but they are clearly doing it on purpose). Do you have an example of a situation this proposal would enable which is not already possible and that we want to avoid by all means?

[astrojuanlu]

actually I am expecting this to be an actual improvement for users

Hard agree. Our YAML-first approach to credentials is weird, given that there are many other methods that are arguably more prevalent, like environment variables, secrets vaults, JSON token files...

Plus we have evidence that users struggle to understand the current approach #1646 (comment)

[datajoely]

I think this is really elegant, I think there is probably a way where this is non-breaking and just an extension too

[idanov]

@Galileo-Galilei The way OmegaConf resolvers work is that they eventually render the value to the dictionary you have provided. That means that in the end result, you get the credentials as part of the key there. The consumer (in this case Kedro) would see no difference between this:

# new proposal
my_data:
    type: pandas.CSVDataset
    filepath: /path/to/data.csv
    credentials: ${credentials: my_credentials}

and this:

# new proposal
my_data:
    type: pandas.CSVDataset
    filepath: /path/to/data.csv
    credentials: s3cre7!

Currently, the latter will not be taken by Kedro and it won't work, because we do the resolving after loading the config. This way we are preventing users from accidentally checking in secrets. If we go the resolver path (which I like personally), we need to make sure that my second example will error out and your example will only work.

[noklam]

@idanov I do not think this is the case even for now, from my vague memory this has been possible for at least 2+ years.

I do not recall any validation against the credentials key, the only path that it leads to error is that passing a str to credentials because Kedro would try to resolve it from credentials.yml, in case it is a dictionary already, Kedro is happy to just take it.

[Galileo-Galilei]

This is a totally valid concern, I did not think about this edge case.

It seems it should be possible to check if the credentials subkey is interpolated with [OmegaConf.is_interpolation method] (https://omegaconf.readthedocs.io/en/2.3_branch/usage.html#omegaconf-is-interpolation) and raise an error if it isn't, but it will likely require #2973 to be completed first (and given the number of references to this issue, it feels like it will become a blocker at one point)

[astrojuanlu]

That’s exactly what I always do: I use credentials as env vars. however, it would still be nice to be able to inject them in a dataset (think db credentials for sql datasets) in some way.

Originally posted by @MatthiasRoels in #1936 (comment)

[datajoely]

O also think a resolver based way of doing creds would be amazing, I had to hack something together to make things work for my SQLFrame experiement.

[m-gris]

Hi Everyone,

As always, a lot of great thoughts & comments. 👍

If I may quickly had some poorly articulated concerns / suggestions that connects this issue to #2829:

AWS Secrets bills on READ ⚠️ 💵 , making it crucial that only credentials that are actually needed are being loaded.

Our problem is compounded by the fact that, in the @after_context_created hook, there is unfortunately no access to tags, pipeline name, namespace ..., i.e data that would help in filtering things out and minimise AWS billing.

Looking very forward to reading your thoughts on this.

[noklam]

The suggestion make sense. We agree this is something that is useful, at the moment we are doing some re-design for DataCatalog, we will start looking at this once we finish the refactoring.

[astrojuanlu]

More evidence that our users are just using environment variables for credentials https://kedro.hall.community/support-lY6wDVhxGXNY/passing-credentials-to-a-node-in-kedro-DrfL3JQdGpPb

(Related to #3979 cc @lordsoffallen)

[astrojuanlu]

More users using env vars for credentials https://kedro.hall.community/support-lY6wDVhxGXNY/using-kubernetes-secrets-in-kedro-pipelines-for-s3-and-external-service-connections-bqgdvd7PZT5J#44b40d5e-cea9-4b5c-b251-35139b3de2ed

[astrojuanlu]

Credential management in Polars is changing 👀 see

• Allow Overriding Object Store Credential Provider pola-rs/polars#18979
• feat: Experimental credential_provider argument for scan_parquet pola-rs/polars#19271
• feat: Add credential provider utility classes for AWS, GCP pola-rs/polars#19297

lf = pl.scan_parquet(
    "s3://...",
    credential_provider=pl.CredentialProviderAWS(profile_name="..."),
)

[datajoely]

There is a real argument that we could steal a lot of these patterns and trust that the Polars team has put a lot of thought into quite a similar problem space with all of the prior-art that has come before it. Kedro predates things such as fsspec so we should not hold our credential management approach up as anything close to sacrosanct

[astrojuanlu]

Moving this to Discussions to continue the conversation there.

[f-vt]

Hi,

This would be very convenient, especially with dynamic credentials generated from Hooks, so I'll be able to pass a default value if the credential doesn't exists.

Today I've found a way by adding a "get_credential_if_available" resolver in an "after_context_created" hook:

class GetCredentialIfAvailableResolver:
    @hook_impl
    def after_context_created(self, context) -> None:
        # This method is called after the Kedro context is created, before catalog creation

        def custom_resolver(context, credential_name):
            # return the credential if it exists, else unset the credential by returning None
            try:
                return context.config_loader["credentials"][credential_name]
            except KeyError:
                return None
        # registering the resolver
        OmegaConf.register_new_resolver("get_credential_if_available", lambda x: custom_resolver(context, x))

That I then uses in it my data catalog, like this:

master_table:
  type: pandas.ExcelDataset
  filepath:  "${runtime_params:mastertable-file-blob, /data/master_table.xlsx}"
  #save_args:
  credentials: ${get_credential_if_available:mastertable-file-blob_credentials}

This is useful as my "mastertable-file-blob_credentials" is generated only if the passed parameter is an abfs path --params=mastertable-file-blob="abfs://account_name.dfs.core.windows.net/container_name/file/path.xlsx".

So now I don't get any error if the credential doesn't exists (so when I run the pipeline locally for developing and don't pass an abfs:// path with the kedro commandline parameter, that is just saving to local file to the path which is the default value).

The hook for generating the dynamic credential if it helps understanding:

class AzureBlobDefaultAzureCredentialsBlobPathsParametersCredentialsGenerator:
    # This class automatically detects STRING values parameters which contains a blob url and annd credentials for them, with the same name than the parameter but appended with _credentials
    # it uses Azure Default Credentials, wrapped to async for fsspec

    @hook_impl

    def after_context_created(self, context) -> None:

        class AsyncDefaultAzureCredentialWrapper(AsyncTokenCredential):
            """
            This class wraps the synchronous DefaultAzureCredential to provide an asynchronous interface.
            It allows the use of DefaultAzureCredential with fsspec, which requires asynchronous credentials.
            The close method is overridden to be a no-op to prevent issues when fsspec tries to close the credential.
            Note from Francis Viallet: I posted the answer to the fsspec's git issue related to this: https://github.com/fsspec/adlfs/issues/431
            """

            def __init__(self, **kwargs):
                # Initialize the synchronous DefaultAzureCredential
                self._credential = DefaultAzureCredential(**kwargs)

            async def get_token(self, *scopes, **kwargs):
                #Asynchronously get a token for the specified scopes: this method runs the synchronous get_token method in an executor to avoid blocking.
                loop = asyncio.get_event_loop()
                token = await loop.run_in_executor(None, self._credential.get_token, *scopes, **kwargs)
                return AccessToken(token.token, token.expires_on)

            async def close(self):
                #No-op close method to prevent issues when fsspec tries to close the credential.
                pass

        # Create the async credential from the sync one, with the no-op close
        credential = AsyncDefaultAzureCredentialWrapper(exclude_interactive_browser_credential=False)

        # This method is called after the Kedro context is created, before catalog creation (which we need as we will use the credentials in the catalogs): https://docs.kedro.org/en/latest/hooks/introduction.html
        
        # Filter parametetrs to get ones with azure datalake/blob fsspec's compliant path values only (pandas datasets uses fsspec derrière)
        blob_paths_parameters = {
            k: v for k, v in context.params.items() 
            if isinstance(v, str) 
            and (v.startswith('abfs://') or v.startswith('az://') or v.startswith('adl://')) 
            and ('.blob.core.windows.net/' in v or '.dfs.core.windows.net/' in v)
        }

        if len(blob_paths_parameters) > 0:
            # Initialize the DefaultAzureCredential if needed (so if parameters containing blob path were found)
            # We disable the exclude_interactive_browser_credential so when this class will ask user to authenticate in last resort if none of the previous methods works: useful when running the code on developer's Computer 
            #from azure.identity import DefaultAzureCredential # import it there because fsspec needs to uses the aio (async) version where others uses the regular one (https://github.com/fsspec/adlfs/issues/431)
            #credential = DefaultAzureCredential(exclude_interactive_browser_credential=False)

            # Iterate found parameters 
            for key, blob_path in blob_paths_parameters.items():
            
                # extract account name from blob path
                if '@' in blob_path:
                    account_name = blob_path.split('@')[-1].split('.')[0]
                else:
                    url_parsed = urlparse(blob_path)
                    account_name = url_parsed.netloc.split('.')[0]

                # Adding the credentials to the Kedro context configuration (will be added to setting.py to made them available to all the catalog/pipelines : from vmx_preprocessing.hooks import AzureSQLDefaultAzureCredentials, AzureSharepointDefaultAzureCredentials ; HOOKS = (AzureSQLDefaultAzureCredentials(), AzureSharepointDefaultAzureCredentials(),))
                context.config_loader["credentials"] = {
                    **context.config_loader["credentials"],# Keep existing credentials in the context to not lose them when overwriting the credentials object

                    f'{key}_credentials' : {
                        'account_name': account_name,
                        'credential': credential,
                        'anon': False
                    }
                }

[astrojuanlu]

Today I found something interesting:

credentials has magical handling:

kedro/kedro/io/catalog_config_resolver.py

Line 18 in f6ecca5

CREDENTIALS_KEY = "credentials"

...same as filepath #2942

...but this is not transparent for users! I just helped a user that had a custom dataset and entries like these:

# catalog.yml
ds:
  type: my.custom.Dataset
  url: db_url

# credentials.yml
db_url: https://...

They expected this to work, but they were getting overwhelming errors essentially boiling down to the string "db_url" not being a valid connection URL.

They also had the perception that it was not okay to use environment variables.

So in the end I saw two options: either they changed their catalog entries and custom dataset source code to abide by the credentials magic management, or they stored the secret values in globals.yml and used the ${globals:} resolver.

(After which they asked, "why can't I do the same with keys stored in parameters.yml?", and I don't think it's possible but couldn't locate the conversation in this issue tracker)