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: Optional[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 is not None: | |||||
query_filters.append("enabled = %s") | |||||
query_params.append(enabled) | |||||
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 |