Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/db.py
Show First 20 Lines • Show All 519 Lines • ▼ Show 20 Lines | origin_visit_select_cols = [ | ||||
"ov.visit", | "ov.visit", | ||||
"ov.date", | "ov.date", | ||||
"ov.type AS type", | "ov.type AS type", | ||||
"ovs.status", | "ovs.status", | ||||
"ovs.metadata", | "ovs.metadata", | ||||
"ovs.snapshot", | "ovs.snapshot", | ||||
] | ] | ||||
def _make_origin_visit_status(self, row: Tuple[Any]) -> Optional[Dict[str, Any]]: | origin_visit_status_select_cols = [ | ||||
"o.url AS origin", | |||||
"ovs.visit", | |||||
"ovs.date", | |||||
"ovs.status", | |||||
"ovs.snapshot", | |||||
"ovs.metadata", | |||||
] | |||||
def _make_origin_visit_status( | |||||
self, row: Optional[Tuple[Any]] | |||||
) -> Optional[Dict[str, Any]]: | |||||
"""Make an origin_visit_status dict out of a row | """Make an origin_visit_status dict out of a row | ||||
""" | """ | ||||
if not row: | if not row: | ||||
return None | return None | ||||
return dict(zip(self.origin_visit_status_cols, row)) | return dict(zip(self.origin_visit_status_cols, row)) | ||||
def origin_visit_status_get_latest( | def origin_visit_status_get_latest( | ||||
self, origin: str, visit: int, cur=None | self, | ||||
origin_url: str, | |||||
visit: int, | |||||
allowed_statuses: Optional[List[str]] = None, | |||||
require_snapshot: bool = False, | |||||
cur=None, | |||||
) -> Optional[Dict[str, Any]]: | ) -> Optional[Dict[str, Any]]: | ||||
"""Given an origin visit id, return its latest origin_visit_status | """Given an origin visit id, return its latest origin_visit_status | ||||
""" | """ | ||||
cols = self.origin_visit_status_cols | |||||
cur = self._cursor(cur) | cur = self._cursor(cur) | ||||
cur.execute( | |||||
f"SELECT {', '.join(cols)} " | query_parts = [ | ||||
f"FROM origin_visit_status ovs " | "SELECT %s" % ", ".join(self.origin_visit_status_select_cols), | ||||
f"INNER JOIN origin o on o.id=ovs.origin " | "FROM origin_visit_status ovs ", | ||||
f"WHERE o.url=%s AND ovs.visit=%s" | "INNER JOIN origin o ON o.id = ovs.origin", | ||||
f"ORDER BY ovs.date DESC LIMIT 1", | ] | ||||
(origin, visit), | query_parts.append("WHERE o.url = %s") | ||||
) | query_params: List[Any] = [origin_url] | ||||
query_parts.append("AND ovs.visit = %s") | |||||
query_params.append(visit) | |||||
if require_snapshot: | |||||
query_parts.append("AND ovs.snapshot is not null") | |||||
if allowed_statuses: | |||||
query_parts.append("AND ovs.status IN %s") | |||||
query_params.append(tuple(allowed_statuses)) | |||||
query_parts.append("ORDER BY ovs.date DESC LIMIT 1") | |||||
query = "\n".join(query_parts) | |||||
cur.execute(query, tuple(query_params)) | |||||
row = cur.fetchone() | row = cur.fetchone() | ||||
return self._make_origin_visit_status(row) | return self._make_origin_visit_status(row) | ||||
def origin_visit_get_all(self, origin_id, last_visit=None, limit=None, cur=None): | def origin_visit_get_all(self, origin_id, last_visit=None, limit=None, cur=None): | ||||
"""Retrieve all visits for origin with id origin_id. | """Retrieve all visits for origin with id origin_id. | ||||
Args: | Args: | ||||
origin_id: The occurrence's origin | origin_id: The occurrence's origin | ||||
▲ Show 20 Lines • Show All 661 Lines • Show Last 20 Lines |