Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/interface.py
# Copyright (C) 2015-2020 The Software Heritage developers | # Copyright (C) 2015-2020 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, Optional | from typing import Any, Dict, Optional | ||||
from swh.core.api import remote_api_endpoint | from swh.core.api import remote_api_endpoint | ||||
from swh.scheduler.model import Lister | |||||
class SchedulerInterface: | class SchedulerInterface: | ||||
@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. | ||||
Args: | Args: | ||||
task_type (dict): a dictionary with the following keys: | task_type (dict): a dictionary with the following keys: | ||||
▲ Show 20 Lines • Show All 226 Lines • ▼ Show 20 Lines | def delete_archived_tasks(self, task_ids): | ||||
""" | """ | ||||
... | ... | ||||
@remote_api_endpoint("task_run/get") | @remote_api_endpoint("task_run/get") | ||||
def get_task_runs(self, task_ids, limit=None): | def get_task_runs(self, task_ids, limit=None): | ||||
"""Search task run for a task id""" | """Search task run for a task id""" | ||||
... | ... | ||||
@remote_api_endpoint("lister/get_or_create") | |||||
def get_or_create_lister( | |||||
self, name: str, arguments: Optional[Dict[str, str]] = None | |||||
) -> Lister: | |||||
"""Retrieve (or create) the lister with the given name and given arguments.""" | |||||
... | |||||
@remote_api_endpoint("priority_ratios/get") | @remote_api_endpoint("priority_ratios/get") | ||||
def get_priority_ratios(self): | def get_priority_ratios(self): | ||||
... | ... |