nrp_cmd.config

 28@define(kw_only=True)
 29class Config:
 30    """The configuration of the NRP client as stored in the configuration file."""
 31
 32    repositories: list[RepositoryConfig] = field(factory=list)
 33    """Locally known repositories."""
 34
 35    default_alias: Optional[str] = None
 36    """The alias of the default repository"""
 37
 38    per_directory_variables: bool = True
 39    """Whether to load variables from a .nrp directory in the current directory.
 40       If set to False, the variables are loaded from the global configuration file
 41       located in ~/.nrp/variables.json.
 42    """
 43
 44    _config_file_path: Optional[Path] = None
 45    """The path from which the config file was loaded."""
 46
 47    @classmethod
 48    def from_file(cls, config_file_path: Optional[Path] = None) -> Self:
 49        """Load the configuration from a file."""
 50        if not config_file_path:
 51            if "NRP_CMD_CONFIG_PATH" in os.environ:
 52                config_file_path = Path(os.environ["NRP_CMD_CONFIG_PATH"])
 53            else:
 54                config_file_path = Path.home() / ".nrp" / "invenio-config.json"
 55
 56        if config_file_path.exists() and config_file_path.stat().st_size > 0:
 57            ret = converter.structure(
 58                json.loads(config_file_path.read_text(encoding="utf-8")), cls
 59            )
 60        else:
 61            ret = cls()
 62        ret._config_file_path = config_file_path
 63        return ret
 64
 65    def save(self, path: Optional[Path] = None) -> None:
 66        """Save the configuration to a file, creating parent directory if needed."""
 67        if path:
 68            self._config_file_path = path
 69        else:
 70            path = self._config_file_path
 71        assert path is not None
 72        path.parent.mkdir(parents=True, exist_ok=True)
 73        path.write_text(json.dumps(converter.unstructure(self), indent=4))
 74
 75    #
 76    # Repository management
 77    #
 78
 79    def get_repository(self, alias: str) -> RepositoryConfig:
 80        """Get a repository by its alias."""
 81        for repo in self.repositories:
 82            if repo.alias == alias:
 83                if not repo.enabled:
 84                    raise ValueError(f"Repository with alias '{alias}' is disabled")
 85                return repo
 86        raise KeyError(f"Repository with alias '{alias}' not found")
 87
 88    @property
 89    def default_repository(self) -> RepositoryConfig:
 90        """Get the default repository."""
 91        if self.default_alias is None:
 92            raise ValueError("Default repository not set")
 93        return self.get_repository(self.default_alias)
 94
 95    def add_repository(self, repository: RepositoryConfig) -> None:
 96        """Add a repository to the configuration."""
 97        self.repositories.append(repository)
 98
 99    def remove_repository(self, repository: RepositoryConfig | str) -> None:
100        """Remove a repository from the configuration."""
101        if isinstance(repository, str):
102            repository = self.get_repository(repository)
103        self.repositories.remove(repository)
104
105    def set_default_repository(self, alias: str) -> None:
106        """Set the default repository by its alias."""
107        try:
108            next(repo for repo in self.repositories if repo.alias == alias)
109        except StopIteration:
110            raise ValueError(f"Repository with alias '{alias}' not found") from None
111        self.default_alias = alias
112
113    def get_repository_from_url(self, record_url: str | URL) -> RepositoryConfig:
114        """Get the repository configuration for a given record URL.
115
116        If there is no repository configuration for the given URL, a dummy
117        repository configuration is returned.
118        """
119        record_url = URL(record_url)
120        repository_root_url = record_url.with_path("/")
121        for repository in self.repositories:
122            if not repository.enabled:
123                continue
124            if repository.url == repository_root_url:
125                return repository
126        # return a dummy repository configuration
127        return RepositoryConfig(
128            alias=str(repository_root_url),
129            url=repository_root_url,
130            info=None,
131        )
132
133    def find_repository(self, repository: URL | str) -> RepositoryConfig:
134        """Find a repository configuration by its URL or alias name."""
135
136        # try alias first
137        if isinstance(repository, str):    
138            with contextlib.suppress(KeyError):
139                return self.get_repository(repository)
140        # if not successful, use the url to get/configure a repository
141        return self.get_repository_from_url(repository)
142
143    def load_variables(self) -> Variables:
144        """Load the global variables from the configuration file."""
145        if self.per_directory_variables:
146            return Variables.from_file(Path.cwd() / ".nrp" / "variables.json")
147        return Variables.from_file()

 19@define(kw_only=True)
 20class RepositoryConfig:
 21    """Configuration of the repository."""
 22
 23    alias: str
 24    """The local alias of the repository."""
 25
 26    url: URL
 27    """The api URL of the repository, usually something like https://repository.org/api."""
 28
 29    token: Optional[str] = None
 30    """Bearer token"""
 31
 32    verify_tls: bool = True
 33    """Verify the TLS certificate in https"""
 34
 35    retry_count: int = 10
 36    """Number of times idempotent operations will be retried if something goes wrong."""
 37
 38    retry_after_seconds: int = 10
 39    """If server does not suggest anything else, retry after this interval in seconds"""
 40
 41    info: RepositoryInfo | None = None
 42    """Cached repository info"""
 43
 44    enabled: bool = True
 45    
 46    class Config:  # noqa
 47        extra = "forbid"
 48
 49    @property
 50    def well_known_repository_url(self) -> URL:
 51        """Return URL to the well-known repository endpoint."""
 52        return self.url / ".well-known" / "repository/"
 53
 54    def search_url(self, model: str | None) -> URL:
 55        """Return URL to search for published records within a model."""
 56        assert self.info is not None
 57        model = model or self._default_model_name
 58        if model:
 59            return self.info.models[model].links.published
 60        return self.info.links.records
 61
 62    def user_search_url(self, model: str | None) -> URL:
 63        """Return URL to search for user's records within a model."""
 64        assert self.info is not None
 65        model = model or self._default_model_name
 66        if model:
 67            user_records = self.info.models[model].links.user_records
 68            if user_records is None:
 69                return self.search_url(model)
 70            return user_records
 71        return self.info.links.user_records
 72
 73    def create_url(self, model: str | None) -> URL:
 74        """Return URL to create a new record within a model."""
 75        assert self.info is not None
 76        model = model or self._default_model_name
 77        if model:
 78            return self.info.models[model].links.api
 79        return self.info.links.records
 80
 81    def read_url(self, model: str | None, record_id: str | URL) -> URL:
 82        """Return URL to a published record within a model."""
 83        assert self.info is not None
 84        if isinstance(record_id, URL):
 85            return record_id
 86        if record_id.startswith("https://"):
 87            return URL(record_id)
 88        model = model or self._default_model_name
 89        if model:
 90            return self.info.models[model].links.api / record_id
 91        return self.info.links.records / record_id
 92
 93    def user_read_url(self, model: str | None, record_id: str | URL) -> URL:
 94        """Return URL to a draft record within a model."""
 95        assert self.info is not None
 96        if isinstance(record_id, URL):
 97            return record_id
 98        if record_id.startswith("https://"):
 99            return URL(record_id)
100        model = model or self._default_model_name
101        if model:
102            return self.info.models[model].links.api / record_id / "draft"
103        return self.info.links.records / record_id / "draft"
104
105    @property
106    def requests_url(self) -> URL:
107        """Return URL to the requests endpoint."""
108        assert self.info is not None
109        return self.info.links.requests
110
111    @property
112    def _default_model_name(self) -> str | None:
113        """Return the default model name if there is only one model in the repository."""
114        if self.info and len(self.info.models) == 1:
115            return next(iter(self.info.models))
116        return None

46    class Config:  # noqa
47        extra = "forbid"