Differential D165 Diff 512 swh/lister/core/tasks.py

Changeset View

Standalone View

swh/lister/core/tasks.py

This file was added.

				# Copyright (C) 2017 the Software Heritage developers
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information

				import random
				from celery import group
				from swh.scheduler.task import Task
				import abc
				from .abstractattribute import AbstractAttribute

				zackUnsubmitted Done Inline Actions This module looks like a good place where to briefly describe the different listing "discplines" we support, one per task. zack: This module looks like a good place where to briefly describe the different listing…
				zackUnsubmitted Done Inline Actions This is still pending — as shown by the TODO marked right below this hunk — right? Do you prefer to merge leaving that pending or what? zack: This is still pending — as shown by the TODO marked right below this hunk — right? Do you…
				fiendishAuthorUnsubmitted Done Inline Actions I don't _want_ to leave it pending. I just don't have a good description of tasks to put in there. :) fiendish: I don't _want_ to leave it pending. I just don't have a good description of tasks to put in…

				class ListerTaskBase(abc.ABC, Task):
				GROUP_SPLIT = 10000
				NAME = AbstractAttribute
				API_URL = AbstractAttribute
				Lister = AbstractAttribute
				task_queue = AbstractAttribute

				def new_lister(self):
				return self.Lister(self.NAME, self.API_URL)

				@abc.abstractmethod
				def run(self):
				pass


				class IncrementalLister(ListerTaskBase):
				def run(self):
				lister = self.new_lister()
				lister.fetch_pages(min_index=lister.last_repo_index(), max_index=None)


				class RangeLister(ListerTaskBase):
				def run(self, start, end):
				lister = self.new_lister()
				lister.fetch_pages(min_index=start, max_index=end)


				class FullLister(ListerTaskBase):
				def run(self):
				lister = self.new_lister()
				ranges = lister.partition_repo_indices(self.GROUP_SPLIT)
				random.shuffle(ranges)
				range_task = RangeLister()
				group(range_task.s(min, max) for min, max in ranges)()