Differential D4880 Diff 17360 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
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,		OriginVisitStats,
PaginatedListedOriginList,		PaginatedListedOriginList,
		SchedulerMetrics,
)		)


@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 309 Lines • ▼ Show 20 Lines	) -> None:
...		...

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

		@remote_api_endpoint("scheduler_metrics/update")
		def update_metrics(
		self,
		lister_id: Optional[UUID] = None,
		timestamp: Optional[datetime.datetime] = None,
		) -> List[SchedulerMetrics]:
		"""Update the performance metrics of this scheduler instance.

		Returns the updated metrics.

		Args:
		lister_id: if passed, update the metrics only for this lister instance
		timestamp: if passed, the date at which we're updating the metrics,
		defaults to the database NOW()
		"""
		...

		@remote_api_endpoint("scheduler_metrics/get")
		def get_metrics(
		self, lister_id: Optional[UUID] = None, visit_type: Optional[str] = None
		) -> List[SchedulerMetrics]:
		"""Retrieve the performance metrics of this scheduler instance.

		Args:
		lister_id: filter the metrics for this lister instance only
		visit_type: filter the metrics for this visit type only
		"""
		...