Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/in_memory.py
| Show First 20 Lines • Show All 810 Lines • ▼ Show 20 Lines | def snapshot_get_latest(self, origin, allowed_statuses=None): | ||||
| branches. | branches. | ||||
| """ | """ | ||||
| if isinstance(origin, int): | if isinstance(origin, int): | ||||
| origin = self.origin_get({'id': origin})['url'] | origin = self.origin_get({'id': origin})['url'] | ||||
| visit = self.origin_visit_get_latest( | visit = self.origin_visit_get_latest( | ||||
| origin, allowed_statuses=allowed_statuses, require_snapshot=True) | origin, allowed_statuses=allowed_statuses, require_snapshot=True) | ||||
| if visit and visit['snapshot']: | if visit and visit['snapshot']: | ||||
| return self.snapshot_get(visit['snapshot']) | snapshot = self.snapshot_get(visit['snapshot']) | ||||
| if not snapshot: | |||||
| raise ValueError( | |||||
| 'last origin visit references an unknown snapshot') | |||||
| return snapshot | |||||
| def snapshot_count_branches(self, snapshot_id, db=None, cur=None): | def snapshot_count_branches(self, snapshot_id, db=None, cur=None): | ||||
| """Count the number of branches in the snapshot with the given id | """Count the number of branches in the snapshot with the given id | ||||
| Args: | Args: | ||||
| snapshot_id (bytes): identifier of the snapshot | snapshot_id (bytes): identifier of the snapshot | ||||
| Returns: | Returns: | ||||
| ▲ Show 20 Lines • Show All 530 Lines • ▼ Show 20 Lines | def origin_visit_get_latest( | ||||
| """ | """ | ||||
| origin = self.origin_get({'url': origin})['id'] | origin = self.origin_get({'url': origin})['id'] | ||||
| visits = self._origin_visits[origin-1] | visits = self._origin_visits[origin-1] | ||||
| if allowed_statuses is not None: | if allowed_statuses is not None: | ||||
| visits = [visit for visit in visits | visits = [visit for visit in visits | ||||
| if visit['status'] in allowed_statuses] | if visit['status'] in allowed_statuses] | ||||
| if require_snapshot: | if require_snapshot: | ||||
| visits = [visit for visit in visits | visits = [visit for visit in visits | ||||
| if visit['snapshot'] | if visit['snapshot']] | ||||
| and visit['snapshot'] in self._snapshots] | |||||
| return max(visits, key=lambda v: (v['date'], v['visit']), default=None) | return max(visits, key=lambda v: (v['date'], v['visit']), default=None) | ||||
| def person_get(self, person): | def person_get(self, person): | ||||
| """Return the persons identified by their ids. | """Return the persons identified by their ids. | ||||
| Args: | Args: | ||||
| person: array of ids. | person: array of ids. | ||||
| ▲ Show 20 Lines • Show All 235 Lines • Show Last 20 Lines | |||||