Fix token retrieval for invenio

Do not force to use the Vault and rely on the config template value.
This commit is contained in:
davelopez
2024-04-16 12:54:10 +02:00
parent 346b85fe23
commit ae64de668d
+11 -23
View File
@@ -1,6 +1,5 @@
import logging
from typing import (
cast,
List,
NamedTuple,
Optional,
@@ -138,11 +137,10 @@ class RDMFilesSource(BaseFilesSource):
def __init__(self, **kwd: Unpack[FilesSourceProperties]):
props = self._parse_common_config_opts(kwd)
base_url = props.get("url", None)
base_url = props.get("url")
if not base_url:
raise Exception("URL for RDM repository must be provided in configuration")
self._repository_url = base_url
self._token = props.get("token", None)
self._props = props
self._repository_interactor = self.get_repository_interactor(base_url)
@@ -150,10 +148,6 @@ class RDMFilesSource(BaseFilesSource):
def repository(self) -> RDMRepositoryInteractor:
return self._repository_interactor
@property
def token(self) -> Optional[str]:
return self._token if self._token and not self._token.startswith("$") else None
def get_repository_interactor(self, repository_url: str) -> RDMRepositoryInteractor:
"""Returns an interactor compatible with the given repository URL.
@@ -190,25 +184,19 @@ class RDMFilesSource(BaseFilesSource):
def get_record_id_from_path(self, source_path: str) -> str:
return self.parse_path(source_path, record_id_only=True).record_id
def _serialization_props(self, user_context: OptionalUserContext = None) -> RDMFilesSourceProperties:
def _serialization_props(self, user_context: OptionalUserContext = None):
effective_props = {}
for key, val in self._props.items():
effective_props[key] = self._evaluate_prop(val, user_context=user_context)
effective_props["url"] = self._repository_url
effective_props["token"] = self.safe_get_authorization_token(user_context)
return cast(RDMFilesSourceProperties, effective_props)
return effective_props
def get_authorization_token(self, user_context: OptionalUserContext) -> str:
token = self.token
if not token and user_context:
vault = user_context.user_vault if user_context else None
token = vault.read_secret(f"preferences/{self.id}/token") if vault else None
if token is None:
raise AuthenticationRequired(f"No authorization token provided in user's settings for '{self.label}'")
token = None
if user_context:
effective_props = self._serialization_props(user_context)
token = effective_props.get("token")
if not token:
raise AuthenticationRequired(
f"Please provide a personal access token in your user's preferences for '{self.label}'"
)
return token
def safe_get_authorization_token(self, user_context: OptionalUserContext) -> Optional[str]:
try:
return self.get_authorization_token(user_context)
except AuthenticationRequired:
return None