Source code for renku.core.util.metadata

# Copyright Swiss Data Science Center (SDSC). A partnership between
# École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
"""Helpers functions for metadata management/parsing."""

import fnmatch
import os
import re
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

from packaging.version import Version

from renku.core import errors
from renku.core.config import get_value, set_value
from renku.core.migration.utils import OLD_METADATA_PATH
from renku.core.util import communication
from renku.core.util.os import is_subpath

    from renku.core.dataset.providers.api import ProviderCredentials
    from renku.domain_model.provenance.agent import Person

[docs]def construct_creators(creators: List[Union[dict, str]], ignore_email=False) -> Tuple[List["Person"], List[str]]: """Parse input and return a list of Person.""" creators = creators or [] people = [] no_email_warnings = [] for creator in creators: person, no_email_warning = construct_creator(creator, ignore_email=ignore_email) if person: people.append(person) if no_email_warning: no_email_warnings.append(no_email_warning) return people, no_email_warnings
[docs]def construct_creator(creator: Union[dict, str], ignore_email) -> Tuple[Optional["Person"], Optional[str]]: """Parse input and return an instance of Person.""" from renku.domain_model.provenance.agent import Person if not creator: return None, None if isinstance(creator, str): person = Person.from_string(creator) elif isinstance(creator, dict): person = Person.from_dict(creator) else: raise errors.ParameterError("Invalid creator type") message = 'A valid format is "Name <email> [affiliation]"' if not # pragma: no cover raise errors.ParameterError(f'Name is invalid: "{creator}".\n{message}') if not if not ignore_email: # pragma: no cover raise errors.ParameterError(f'Email is invalid: "{creator}".\n{message}') else: no_email_warning = str(creator) else: no_email_warning = None return person, no_email_warning
[docs]def is_external_file(path: Union[Path, str], project_path: Path): """Checks if a path is an external file.""" path = project_path / path if not path.is_symlink() or not is_subpath(path=path, base=project_path): return False pointer = os.readlink(path) return str(os.path.join(RENKU_HOME, POINTERS)) in pointer
[docs]def is_linked_file(path: Union[Path, str], project_path: Path) -> bool: """Return True if a dataset file is a linked file.""" path = project_path / path if not path.is_symlink() or not is_subpath(path=path.resolve(), base=project_path): return False pointer = os.readlink(path) return os.path.join(RENKU_HOME, POINTERS) in pointer
[docs]def read_renku_version_from_dockerfile(path: Optional[Union[Path, str]] = None) -> Optional[Version]: """Read RENKU_VERSION from the content of path if a valid version is available.""" from renku.domain_model.project_context import project_context path = Path(path) if path else project_context.dockerfile_path if not path.exists(): return None m ="^\s*ARG RENKU_VERSION=(\d+\.\d+\.\d+\S*)$", path.read_text(), flags=re.MULTILINE) if not m: return None try: return Version( except ValueError: return None
[docs]def replace_renku_version_in_dockerfile(dockerfile_content: str, version: str) -> str: """Replace Renku version in the Dockerfile.""" return re.sub( r"^\s*ARG RENKU_VERSION=(\d+\.\d+\.\d+\S*)$", f"ARG RENKU_VERSION={version}", dockerfile_content, flags=re.MULTILINE, )
[docs]def make_project_temp_dir(project_path: Path) -> Path: """Create a temporary directory inside project's temp path.""" base = project_path / RENKU_HOME / RENKU_TMP base.mkdir(parents=True, exist_ok=True) return Path(tempfile.mkdtemp(dir=base))
[docs]def store_credentials(section: str, key: str, value: str) -> None: """Write provider's credentials.""" section, key = _get_section_and_key(section=section, key=key) set_value(section=section, key=key, value=value, global_only=True)
[docs]def read_credentials(section: str, key: str) -> Optional[str]: """Read provider's credentials.""" section, key = _get_section_and_key(section=section, key=key) return get_value(section=section, key=key)
[docs]def get_canonical_key(key: str) -> str: """Make a consistent configuration key.""" return key.replace(" ", "-").lower()
def _get_section_and_key(section: str, key: str) -> Tuple[str, str]: section = "http" if section.lower() == "renku" else section return section, get_canonical_key(key)
[docs]def prompt_for_credentials(provider_credentials: "ProviderCredentials") -> None: """Prompt for provider credentials if needed and update and store them.""" if not provider_credentials: return prompt_to_store = False for key in provider_credentials.get_credentials_names_with_no_value(): prompt_to_store = True value = communication.prompt(f"Enter a value for '{key}'", type=str, default="") provider_credentials[key] = value if prompt_to_store: if communication.confirm("Store credentials?", default=True):
[docs]def is_protected_path(path: Path) -> bool: """Checks if a path is a protected path.""" from renku.domain_model.project_context import project_context try: path_in_repo = str(path.relative_to(project_context.path)) except ValueError: return False for protected_path in RENKU_PROTECTED_PATHS: if fnmatch.fnmatch(path_in_repo, protected_path): return True return False
[docs]def is_renku_project() -> bool: """Check if repository is a renku project.""" from renku.domain_model.project_context import project_context try: return project_context.project is not None except ValueError: # NOTE: Error in loading due to an older schema return project_context.metadata_path.joinpath(OLD_METADATA_PATH).exists()