Differential D3647 Diff 12838 swh/web/common/service.py

Changeset View

Standalone View

swh/web/common/service.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 Affero General Public License version 3, or any later version		# License: GNU Affero 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 itertools		import itertools
import os		import os
import re		import re

from collections import defaultdict		from collections import defaultdict
from typing import Any, Dict, List, Set, Iterable, Iterator, Optional, Tuple		from typing import Any, Dict, List, Set, Iterable, Iterator, Optional, Tuple

from swh.model import hashutil		from swh.model import hashutil
from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT		from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
		from swh.model.model import OriginVisit
from swh.storage.algos import diff, revisions_walker		from swh.storage.algos import diff, revisions_walker
from swh.storage.algos.origin import origin_get_latest_visit_status		from swh.storage.algos.origin import origin_get_latest_visit_status
from swh.storage.algos.snapshot import snapshot_get_latest		from swh.storage.algos.snapshot import snapshot_get_latest
from swh.vault.exc import NotFoundExc as VaultNotFoundExc		from swh.vault.exc import NotFoundExc as VaultNotFoundExc
from swh.web import config		from swh.web import config
from swh.web.common import converters		from swh.web.common import converters
from swh.web.common import query		from swh.web.common import query
from swh.web.common.exc import BadInputExc, NotFoundExc		from swh.web.common.exc import BadInputExc, NotFoundExc
▲ Show 20 Lines • Show All 831 Lines • ▼ Show 20 Lines	def stat_counters():
Returns:		Returns:
A dict mapping textual labels to integer values.		A dict mapping textual labels to integer values.
"""		"""
return storage.stat_counters()		return storage.stat_counters()


def _lookup_origin_visits(		def _lookup_origin_visits(
origin_url: str, last_visit: Optional[int] = None, limit: int = 10		origin_url: str, last_visit: Optional[int] = None, limit: int = 10
) -> Iterator[Dict[str, Any]]:		) -> Iterator[OriginVisit]:
"""Yields the origin origins' visits.		"""Yields the origin origins' visits.

Args:		Args:
origin_url (str): origin to list visits for		origin_url (str): origin to list visits for
last_visit (int): last visit to lookup from		last_visit (int): last visit to lookup from
limit (int): Number of elements max to display		limit (int): Number of elements max to display

Yields:		Yields:
Dictionaries of origin_visit for that origin		OriginVisit for that origin

"""		"""
limit = min(limit, MAX_LIMIT)		limit = min(limit, MAX_LIMIT)
for visit in storage.origin_visit_get(		page_token: Optional[str]
origin_url, last_visit=last_visit, limit=limit		if last_visit is not None:
):		page_token = str(last_visit)
visit["origin"] = origin_url		else:
yield visit		page_token = None
		visit_page = storage.origin_visit_get(
		origin_url, page_token=page_token, limit=limit
		)
		yield from visit_page.results


def lookup_origin_visits(		def lookup_origin_visits(
origin: str, last_visit: Optional[int] = None, per_page: int = 10		origin: str, last_visit: Optional[int] = None, per_page: int = 10
) -> Iterator[OriginVisitInfo]:		) -> Iterator[OriginVisitInfo]:
"""Yields the origin origins' visits.		"""Yields the origin origins' visits.

Args:		Args:
origin: origin to list visits for		origin: origin to list visits for

Yields:		Yields:
Dictionaries of origin_visit for that origin		Dictionaries of origin_visit for that origin

"""		"""
visits = _lookup_origin_visits(origin, last_visit=last_visit, limit=per_page)		for visit in _lookup_origin_visits(origin, last_visit=last_visit, limit=per_page):
for visit in visits:		visit_status = storage.origin_visit_status_get_latest(origin, visit.visit)
visit_status = storage.origin_visit_status_get_latest(origin, visit["visit"])		yield converters.from_origin_visit(
yield converters.from_origin_visit({visit, visit_status.to_dict()})		{**visit_status.to_dict(), "type": visit.type}
		)


def lookup_origin_visit_latest(		def lookup_origin_visit_latest(
origin_url: str,		origin_url: str,
require_snapshot: bool = False,		require_snapshot: bool = False,
type: Optional[str] = None,		type: Optional[str] = None,
allowed_statuses: Optional[List[str]] = None,		allowed_statuses: Optional[List[str]] = None,
) -> Optional[OriginVisitInfo]:		) -> Optional[OriginVisitInfo]:
▲ Show 20 Lines • Show All 406 Lines • Show Last 20 Lines