Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/in_memory.py
| Show First 20 Lines • Show All 940 Lines • ▼ Show 20 Lines | ) -> Optional[OriginVisit]: | ||||
| if visit_statuses: # we found visit statuses matching criteria | if visit_statuses: # we found visit statuses matching criteria | ||||
| visit_status = max(visit_statuses, key=lambda vs: (vs.date, vs.visit)) | visit_status = max(visit_statuses, key=lambda vs: (vs.date, vs.visit)) | ||||
| assert visit.origin == visit_status.origin | assert visit.origin == visit_status.origin | ||||
| assert visit.visit == visit_status.visit | assert visit.visit == visit_status.visit | ||||
| return visit | return visit | ||||
| return None | return None | ||||
| def origin_visit_status_get( | |||||
| self, | |||||
| origin: str, | |||||
| visit: int, | |||||
| page_token: Optional[str] = None, | |||||
| order: ListOrder = ListOrder.ASC, | |||||
| limit: int = 10, | |||||
ardumont: @ardumont have a look here... see... that's why it fails ;)
| |||||
| ) -> PagedResult[OriginVisitStatus]: | |||||
| next_page_token = None | |||||
| date_from = None | |||||
| if page_token is not None: | |||||
| date_from = datetime.datetime.fromisoformat(page_token) | |||||
| visit_statuses = sorted( | |||||
| self._origin_visit_statuses.get((origin, visit), []), | |||||
| key=lambda v: v.date, | |||||
| reverse=(order == ListOrder.DESC), | |||||
| ) | |||||
| if date_from is not None: | |||||
| if order == ListOrder.ASC: | |||||
| visit_statuses = [v for v in visit_statuses if v.date >= date_from] | |||||
| elif order == ListOrder.DESC: | |||||
| visit_statuses = [v for v in visit_statuses if v.date <= date_from] | |||||
| # Take one more visit status so we can reuse it as the next page token if any | |||||
| visit_statuses = visit_statuses[: limit + 1] | |||||
| if len(visit_statuses) > limit: | |||||
| # last visit status date is the next page token | |||||
| next_page_token = str(visit_statuses[-1].date) | |||||
| # excluding that visit status from the result to respect the limit size | |||||
| visit_statuses = visit_statuses[:limit] | |||||
| return PagedResult(results=visit_statuses, next_page_token=next_page_token) | |||||
| def origin_visit_status_get_latest( | def origin_visit_status_get_latest( | ||||
| self, | self, | ||||
| origin_url: str, | origin_url: str, | ||||
| visit: int, | visit: int, | ||||
| allowed_statuses: Optional[List[str]] = None, | allowed_statuses: Optional[List[str]] = None, | ||||
| require_snapshot: bool = False, | require_snapshot: bool = False, | ||||
| ) -> Optional[OriginVisitStatus]: | ) -> Optional[OriginVisitStatus]: | ||||
| ori = self._origins.get(origin_url) | ori = self._origins.get(origin_url) | ||||
| ▲ Show 20 Lines • Show All 262 Lines • Show Last 20 Lines | |||||
@ardumont have a look here... see... that's why it fails ;)