Changeset View
Changeset View
Standalone View
Standalone View
swh/web/common/archive.py
| Show First 20 Lines • Show All 226 Lines • ▼ Show 20 Lines | for license_dict in license_dicts: | ||||
| del license_dict["id"] | del license_dict["id"] | ||||
| lic = { | lic = { | ||||
| "id": sha1, | "id": sha1, | ||||
| "facts": license_dicts, | "facts": license_dicts, | ||||
| } | } | ||||
| return converters.from_swh(lic, hashess={"id"}) | return converters.from_swh(lic, hashess={"id"}) | ||||
| def lookup_origin(origin: OriginInfo) -> OriginInfo: | def lookup_origin(origin: OriginInfo, lookup_similar_urls: bool = True) -> OriginInfo: | ||||
| """Return information about the origin matching dict origin. | """Return information about the origin matching dict origin. | ||||
| Args: | Args: | ||||
| origin: origin's dict with 'url' key | origin: origin's dict with 'url' key | ||||
| lookup_similar_urls: if :const:`True`, lookup origin with and | |||||
| without trailing slash in its URL | |||||
| Returns: | Returns: | ||||
| origin information as dict. | origin information as dict. | ||||
| """ | """ | ||||
| origin_urls = [origin["url"]] | origin_urls = [origin["url"]] | ||||
| if origin["url"]: | if origin["url"] and lookup_similar_urls: | ||||
| # handle case when user provided an origin url with a trailing | # handle case when user provided an origin url with a trailing | ||||
| # slash while the url in storage does not have it (e.g. GitHub) | # slash while the url in storage does not have it (e.g. GitHub) | ||||
| if origin["url"].endswith("/"): | if origin["url"].endswith("/"): | ||||
| origin_urls.append(origin["url"][:-1]) | origin_urls.append(origin["url"][:-1]) | ||||
| # handle case when user provided an origin url without a trailing | # handle case when user provided an origin url without a trailing | ||||
| # slash while the url in storage have it (e.g. Debian source package) | # slash while the url in storage have it (e.g. Debian source package) | ||||
| else: | else: | ||||
| origin_urls.append(f"{origin['url']}/") | origin_urls.append(f"{origin['url']}/") | ||||
| ▲ Show 20 Lines • Show All 717 Lines • ▼ Show 20 Lines | |||||
| 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[OriginVisitWithStatuses]: | ) -> Iterator[OriginVisitWithStatuses]: | ||||
| """Yields the origin origins' visits. | """Yields the origin origins' visits. | ||||
| Args: | Args: | ||||
| origin_url (str): origin to list visits for | origin_url: origin to list visits for | ||||
| last_visit (int): last visit to lookup from | last_visit: last visit to lookup from | ||||
| limit (int): Number of elements max to display | limit: Number of elements max to display | ||||
| Yields: | Yields: | ||||
| OriginVisit for that origin | OriginVisit for that origin | ||||
| """ | """ | ||||
| limit = min(limit, MAX_LIMIT) | limit = min(limit, MAX_LIMIT) | ||||
| page_token: Optional[str] | page_token: Optional[str] | ||||
| if last_visit is not None: | if last_visit is not None: | ||||
| Show All 22 Lines | for visit in _lookup_origin_visits(origin, last_visit=last_visit, limit=per_page): | ||||
| yield converters.from_origin_visit(visit.statuses[-1].to_dict()) | yield converters.from_origin_visit(visit.statuses[-1].to_dict()) | ||||
| 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, | ||||
| lookup_similar_urls: bool = True, | |||||
| ) -> Optional[OriginVisitInfo]: | ) -> Optional[OriginVisitInfo]: | ||||
| """Return the origin's latest visit | """Return the origin's latest visit | ||||
| Args: | Args: | ||||
| origin_url: origin to list visits for | origin_url: origin to list visits for | ||||
| type: Optional visit type to filter on (e.g git, tar, dsc, svn, | type: Optional visit type to filter on (e.g git, tar, dsc, svn, | ||||
| hg, npm, pypi, ...) | hg, npm, pypi, ...) | ||||
| allowed_statuses: list of visit statuses considered | allowed_statuses: list of visit statuses considered | ||||
| to find the latest visit. For instance, | to find the latest visit. For instance, | ||||
| ``allowed_statuses=['full']`` will only consider visits that | ``allowed_statuses=['full']`` will only consider visits that | ||||
| have successfully run to completion. | have successfully run to completion. | ||||
| require_snapshot: filter out origins without a snapshot | require_snapshot: filter out origins without a snapshot | ||||
| lookup_similar_urls: if :const:`True`, lookup origin with and | |||||
| without trailing slash in its URL | |||||
| Returns: | Returns: | ||||
| The origin visit info as dict if found | The origin visit info as dict if found | ||||
| """ | """ | ||||
| # check origin existence in the archive | |||||
| origin_url = lookup_origin( | |||||
| OriginInfo(url=origin_url), lookup_similar_urls=lookup_similar_urls | |||||
| )["url"] | |||||
| visit_status = origin_get_latest_visit_status( | visit_status = origin_get_latest_visit_status( | ||||
| storage, | storage, | ||||
| origin_url, | origin_url, | ||||
| type=type, | type=type, | ||||
| allowed_statuses=allowed_statuses, | allowed_statuses=allowed_statuses, | ||||
| require_snapshot=require_snapshot, | require_snapshot=require_snapshot, | ||||
| ) | ) | ||||
| return ( | return ( | ||||
| converters.from_origin_visit(visit_status.to_dict()) if visit_status else None | converters.from_origin_visit(visit_status.to_dict()) if visit_status else None | ||||
| ) | ) | ||||
| def lookup_origin_visit(origin_url: str, visit_id: int) -> OriginVisitInfo: | def lookup_origin_visit( | ||||
| origin_url: str, | |||||
| visit_id: int, | |||||
| lookup_similar_urls: bool = True, | |||||
| ) -> OriginVisitInfo: | |||||
| """Return information about visit visit_id with origin origin. | """Return information about visit visit_id with origin origin. | ||||
| Args: | Args: | ||||
| origin: origin concerned by the visit | origin: origin concerned by the visit | ||||
| visit_id: the visit identifier to lookup | visit_id: the visit identifier to lookup | ||||
| lookup_similar_urls: if :const:`True`, lookup origin with and | |||||
| without trailing slash in its URL | |||||
| Raises: | Raises: | ||||
| NotFoundExc if no origin visit matching the criteria is found | NotFoundExc if no origin visit matching the criteria is found | ||||
| Returns: | Returns: | ||||
| The dict origin_visit concerned | The dict origin_visit concerned | ||||
| """ | """ | ||||
| # check origin existence in the archive | |||||
| origin_url = lookup_origin( | |||||
| OriginInfo(url=origin_url), lookup_similar_urls=lookup_similar_urls | |||||
| )["url"] | |||||
| visit = storage.origin_visit_get_by(origin_url, visit_id) | visit = storage.origin_visit_get_by(origin_url, visit_id) | ||||
| visit_status = storage.origin_visit_status_get_latest(origin_url, visit_id) | visit_status = storage.origin_visit_status_get_latest(origin_url, visit_id) | ||||
| if not visit: | if not visit: | ||||
| raise NotFoundExc( | raise NotFoundExc( | ||||
| f"Origin {origin_url} or its visit with id {visit_id} not found!" | f"Origin {origin_url} or its visit with id {visit_id} not found!" | ||||
| ) | ) | ||||
| return converters.from_origin_visit({**visit_status.to_dict(), "type": visit.type}) | return converters.from_origin_visit({**visit_status.to_dict(), "type": visit.type}) | ||||
| ▲ Show 20 Lines • Show All 384 Lines • Show Last 20 Lines | |||||