Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/git/tasks.py
| # Copyright (C) 2015-2021 The Software Heritage developers | # Copyright (C) 2015-2022 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from typing import Any, Dict, Optional | from typing import Any, Dict | ||||
| from celery import shared_task | from celery import shared_task | ||||
| import dateutil.parser | |||||
| from swh.loader.git.from_disk import GitLoaderFromArchive, GitLoaderFromDisk | from swh.loader.git.from_disk import GitLoaderFromArchive, GitLoaderFromDisk | ||||
| from swh.loader.git.loader import GitLoader | from swh.loader.git.loader import GitLoader | ||||
| @shared_task(name=__name__ + ".UpdateGitRepository") | @shared_task(name=__name__ + ".UpdateGitRepository") | ||||
| def load_git(*, url: str, base_url: Optional[str] = None) -> Dict[str, Any]: | def load_git(**kwargs) -> Dict[str, Any]: | ||||
| """Import a git repository from a remote location""" | """Import a git repository from a remote location""" | ||||
| loader = GitLoader.from_configfile(url=url, base_url=base_url) | loader = GitLoader.from_configfile(**kwargs) | ||||
| return loader.load() | return loader.load() | ||||
| @shared_task(name=__name__ + ".LoadDiskGitRepository") | @shared_task(name=__name__ + ".LoadDiskGitRepository") | ||||
| def load_git_from_dir(*, url: str, directory: str, date: str) -> Dict[str, Any]: | def load_git_from_dir(**kwargs) -> Dict[str, Any]: | ||||
| """Import a git repository from a local repository | """Import a git repository from a local repository""" | ||||
| loader = GitLoaderFromDisk.from_configfile(**kwargs) | |||||
| Import a git repository, cloned in `directory` from `origin_url` at | |||||
| `date`. | |||||
| """ | |||||
| visit_date = dateutil.parser.parse(date) | |||||
| loader = GitLoaderFromDisk.from_configfile( | |||||
| url=url, directory=directory, visit_date=visit_date | |||||
| ) | |||||
| return loader.load() | return loader.load() | ||||
| @shared_task(name=__name__ + ".UncompressAndLoadDiskGitRepository") | @shared_task(name=__name__ + ".UncompressAndLoadDiskGitRepository") | ||||
| def load_git_from_zip(*, url: str, archive_path: str, date: str) -> Dict[str, Any]: | def load_git_from_zip(**kwargs) -> Dict[str, Any]: | ||||
| """Import a git repository from a zip archive | """Import a git repository from a zip archive | ||||
| 1. Uncompress an archive repository in a local and temporary folder | 1. Uncompress an archive repository in a local and temporary folder | ||||
| 2. Load it through the git disk loader | 2. Load it through the git disk loader | ||||
| 3. Clean up the temporary folder | 3. Clean up the temporary folder | ||||
| """ | """ | ||||
| visit_date = dateutil.parser.parse(date) | loader = GitLoaderFromArchive.from_configfile(**kwargs) | ||||
| loader = GitLoaderFromArchive.from_configfile( | |||||
| url=url, archive_path=archive_path, visit_date=visit_date | |||||
| ) | |||||
| return loader.load() | return loader.load() | ||||