Differential D4905 Diff 17518 swh/scheduler/interface.py

Changeset View

Standalone View

swh/scheduler/interface.py

	# Copyright (C) 2015-2021 The Software Heritage developers			# Copyright (C) 2015-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


	import datetime			import datetime
	from typing import Any, Dict, Iterable, List, Optional, Tuple			from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
	from uuid import UUID			from uuid import UUID

	from typing_extensions import Protocol, runtime_checkable			from typing_extensions import Protocol, runtime_checkable

	from swh.core.api import remote_api_endpoint			from swh.core.api import remote_api_endpoint
	from swh.scheduler.model import (			from swh.core.api.classes import PagedResult
	ListedOrigin,			from swh.scheduler.model import ListedOrigin, Lister, OriginVisitStats, SchedulerMetrics
	ListedOriginPageToken,
	Lister,			ListedOriginPageToken = Tuple[str, str]
	OriginVisitStats,
	PaginatedListedOriginList,
	SchedulerMetrics,			class PaginatedListedOriginList(PagedResult[ListedOrigin, ListedOriginPageToken]):
	)			"""A list of listed origins, with a continuation token"""

				def __init__(
				self,
				results: List[ListedOrigin],
				next_page_token: Union[None, ListedOriginPageToken, List[str]],
				):
				parsed_next_page_token: Optional[Tuple[str, str]] = None
				if next_page_token is not None:
				if len(next_page_token) != 2:
				raise TypeError("Expected Tuple[str, str] or list of size 2.")
				parsed_next_page_token = tuple(next_page_token) # type: ignore
				super().__init__(results, parsed_next_page_token)


	@runtime_checkable			@runtime_checkable
	class SchedulerInterface(Protocol):			class SchedulerInterface(Protocol):
	@remote_api_endpoint("task_type/create")			@remote_api_endpoint("task_type/create")
	def create_task_type(self, task_type):			def create_task_type(self, task_type):
	"""Create a new task type ready for scheduling.			"""Create a new task type ready for scheduling.

	▲ Show 20 Lines • Show All 359 Lines • Show Last 20 Lines