Differential D4838 Diff 17151 swh/scheduler/interface.py

Changeset View

Standalone View

swh/scheduler/interface.py

# Copyright (C) 2015-2020 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

from typing import Any, Dict, Iterable, List, Optional		from typing import Any, Dict, Iterable, List, Optional
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.scheduler.model import (
ListedOrigin,		ListedOrigin,
ListedOriginPageToken,		ListedOriginPageToken,
Lister,		Lister,
		OriginVisitStats,
PaginatedListedOriginList,		PaginatedListedOriginList,
)		)


@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):
▲ Show 20 Lines • Show All 293 Lines • ▼ Show 20 Lines	def grab_next_visits(self, count: int, policy: str,) -> List[ListedOrigin]:
This will mark the origins as "being visited" in the listed_origins		This will mark the origins as "being visited" in the listed_origins
table, to avoid scheduling multiple visits to the same origin.		table, to avoid scheduling multiple visits to the same origin.
"""		"""
...		...

@remote_api_endpoint("priority_ratios/get")		@remote_api_endpoint("priority_ratios/get")
def get_priority_ratios(self):		def get_priority_ratios(self):
...		...

		@remote_api_endpoint("visit_stats/upsert")
		def origin_visit_stats_upsert(self, visit_stats: OriginVisitStats) -> None:
		"""Create a new origin visit stats
		"""
		...

		@remote_api_endpoint("visit_stats/get")
		def origin_visit_stats_get(
		self, url: str, visit_type: str
		) -> Optional[OriginVisitStats]:
		"""Retrieve the stats for an origin with a given visit type"""
		...