Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/launchpad/tasks.py
| # Copyright (C) 2017-2020 The Software Heritage developers | # Copyright (C) 2020-2021 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 celery import shared_task | from celery import shared_task | ||||
| from .lister import LaunchpadLister | from .lister import LaunchpadLister | ||||
| @shared_task(name=__name__ + ".ping") | |||||
| def ping(): | |||||
| return "OK" | |||||
| @shared_task(name=__name__ + ".FullLaunchpadLister") | |||||
| def list_launchpad_full(**lister_args): | |||||
| """Full listing of git repositories hosted on Launchpad""" | |||||
| lister = LaunchpadLister.from_configfile(**lister_args) | |||||
| return lister.run().dict() | |||||
| @shared_task(name=__name__ + ".IncrementalLaunchpadLister") | @shared_task(name=__name__ + ".IncrementalLaunchpadLister") | ||||
| def list_launchpad_incremental(threshold, **lister_args): | def list_launchpad_incremental(**lister_args): | ||||
| """Incremental update | """Incremental listing of git repositories hosted on Launchpad""" | ||||
| """ | lister = LaunchpadLister.from_configfile(**lister_args, incremental=True) | ||||
| lister = LaunchpadLister(**lister_args) | return lister.run().dict() | ||||
| return lister.run(max_bound=threshold) | |||||
| @shared_task(name=__name__ + ".FullLaunchpadLister", bind=True) | |||||
| def list_launchpad_full(self, **lister_args): | |||||
| """Full update of Launchpad | |||||
| """ | |||||
| self.log.debug("%s OK, spawned full task" % (self.name)) | |||||
| return list_launchpad_incremental(threshold=None, **lister_args) | |||||
| @shared_task(name=__name__ + ".NewLaunchpadLister", bind=True) | |||||
| def list_launchpad_new(self, **lister_args): | |||||
| """Update new entries of Launchpad | |||||
| """ | |||||
| lister = LaunchpadLister(**lister_args) | |||||
| threshold = lister.db_last_threshold() | |||||
| self.log.debug("%s OK, spawned new task" % (self.name)) | |||||
| return list_launchpad_incremental(threshold=threshold, **lister_args) | |||||