Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/backend.py
| # Copyright (C) 2015-2021 The Software Heritage developers | # Copyright (C) 2015-2022 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 | ||||
| import json | import json | ||||
| import logging | import logging | ||||
| from typing import Any, Dict, Iterable, List, Optional, Tuple, Union | from typing import Any, Dict, Iterable, List, Optional, Tuple, Union | ||||
| ▲ Show 20 Lines • Show All 299 Lines • ▼ Show 20 Lines | ) -> List[ListedOrigin]: | ||||
| return [ListedOrigin(**d) for d in ret] | return [ListedOrigin(**d) for d in ret] | ||||
| @db_transaction() | @db_transaction() | ||||
| def get_listed_origins( | def get_listed_origins( | ||||
| self, | self, | ||||
| lister_id: Optional[UUID] = None, | lister_id: Optional[UUID] = None, | ||||
| url: Optional[str] = None, | url: Optional[str] = None, | ||||
| enabled_only: bool = True, | |||||
| limit: int = 1000, | limit: int = 1000, | ||||
| page_token: Optional[ListedOriginPageToken] = None, | page_token: Optional[ListedOriginPageToken] = None, | ||||
| db=None, | db=None, | ||||
| cur=None, | cur=None, | ||||
| ) -> PaginatedListedOriginList: | ) -> PaginatedListedOriginList: | ||||
| """Get information on the listed origins matching either the `url` or | """Get information on the listed origins matching either the `url` or | ||||
| `lister_id`, or both arguments. | `lister_id`, or both arguments. | ||||
| """ | """ | ||||
| query_filters: List[str] = [] | query_filters: List[str] = [] | ||||
| query_params: List[Union[int, str, UUID, Tuple[UUID, str]]] = [] | query_params: List[Union[int, str, UUID, Tuple[UUID, str]]] = [] | ||||
| if lister_id: | if lister_id: | ||||
| query_filters.append("lister_id = %s") | query_filters.append("lister_id = %s") | ||||
| query_params.append(lister_id) | query_params.append(lister_id) | ||||
| if url is not None: | if url is not None: | ||||
| query_filters.append("url = %s") | query_filters.append("url = %s") | ||||
| query_params.append(url) | query_params.append(url) | ||||
| if enabled_only: | |||||
| query_filters.append("enabled = true") | |||||
| if page_token is not None: | if page_token is not None: | ||||
| query_filters.append("(lister_id, url) > %s") | query_filters.append("(lister_id, url) > %s") | ||||
| # the typeshed annotation for tuple() is too strict. | # the typeshed annotation for tuple() is too strict. | ||||
| query_params.append(tuple(page_token)) # type: ignore | query_params.append(tuple(page_token)) # type: ignore | ||||
| query_params.append(limit) | query_params.append(limit) | ||||
| select_cols = ", ".join(ListedOrigin.select_columns()) | select_cols = ", ".join(ListedOrigin.select_columns()) | ||||
| ▲ Show 20 Lines • Show All 797 Lines • Show Last 20 Lines | |||||