A local representation of a dataset (metadata) Git repository.
| 178 | |
| 179 | |
| 180 | class DatasetRepository: |
| 181 | """A local representation of a dataset (metadata) Git repository.""" |
| 182 | |
| 183 | def __init__(self, |
| 184 | repo_work_dir: str, |
| 185 | dataset_id: str, |
| 186 | revision: Optional[str] = DEFAULT_DATASET_REVISION, |
| 187 | git_token: Optional[str] = None, |
| 188 | git_path: Optional[str] = None, |
| 189 | endpoint: Optional[str] = None, |
| 190 | auth_token: Optional[str] = None): |
| 191 | if auth_token is not None and git_token is None: |
| 192 | import warnings |
| 193 | warnings.warn( |
| 194 | 'DatasetRepository(auth_token=...) is deprecated, ' |
| 195 | 'use DatasetRepository(git_token=...) instead.', |
| 196 | DeprecationWarning, |
| 197 | stacklevel=2) |
| 198 | git_token = auth_token |
| 199 | |
| 200 | if not repo_work_dir or not isinstance(repo_work_dir, str): |
| 201 | raise InvalidParameter('dataset_work_dir must be provided!') |
| 202 | repo_work_dir = repo_work_dir.rstrip('/') |
| 203 | if not repo_work_dir: |
| 204 | raise InvalidParameter('dataset_work_dir can not be root dir!') |
| 205 | if not revision: |
| 206 | raise InvalidParameter( |
| 207 | 'a non-default value of revision cannot be empty.') |
| 208 | |
| 209 | self._endpoint = endpoint |
| 210 | self.dataset_id = dataset_id |
| 211 | self.repo_work_dir = repo_work_dir |
| 212 | self.repo_base_dir = os.path.dirname(repo_work_dir) |
| 213 | self.repo_name = os.path.basename(repo_work_dir) |
| 214 | self.revision = revision |
| 215 | self.git_token = _resolve_git_token(git_token) |
| 216 | |
| 217 | self.git_wrapper = GitCommandWrapper(git_path) |
| 218 | os.makedirs(self.repo_work_dir, exist_ok=True) |
| 219 | self.repo_url = self._get_repo_url(dataset_id) |
| 220 | |
| 221 | def _get_repo_url(self, dataset_id: str) -> str: |
| 222 | endpoint = self._endpoint or get_endpoint() |
| 223 | return f'{endpoint}/datasets/{dataset_id}.git' |
| 224 | |
| 225 | def clone(self) -> str: |
| 226 | """Clone the dataset repo if not already cloned, returning its path.""" |
| 227 | cloned = _clone_if_needed(self.git_wrapper, self.repo_base_dir, |
| 228 | self.repo_name, self.repo_work_dir, |
| 229 | self.repo_url, self.git_token, self.revision) |
| 230 | return self.repo_work_dir if cloned else '' |
| 231 | |
| 232 | def push(self, |
| 233 | commit_message: str, |
| 234 | branch: Optional[str] = DEFAULT_DATASET_REVISION, |
| 235 | force: bool = False): |
| 236 | """Stage all changes, commit, and push to the remote.""" |
| 237 | warnings.warn( |
no outgoing calls
no test coverage detected
searching dependent graphs…