Differential D3645 Diff 12836 swh/deposit/migrations/0018_migrate_swhids.py

Changeset View

Standalone View

swh/deposit/migrations/0018_migrate_swhids.py

Show All 11 Lines
from swh.model.hashutil import hash_to_bytes, hash_to_hex		from swh.model.hashutil import hash_to_bytes, hash_to_hex
from swh.model.identifiers import (		from swh.model.identifiers import (
parse_swhid,		parse_swhid,
swhid,		swhid,
DIRECTORY,		DIRECTORY,
REVISION,		REVISION,
SNAPSHOT,		SNAPSHOT,
)		)
		from swh.model.model import TargetType
from swh.storage import get_storage as get_storage_client		from swh.storage import get_storage as get_storage_client
		from swh.storage.interface import ListOrder


SWH_PROVIDER_URL = "https://www.softwareheritage.org"		SWH_PROVIDER_URL = "https://www.softwareheritage.org"


logger = logging.getLogger(__name__)		logger = logging.getLogger(__name__)


Show All 39 Lines	def get_storage() -> Optional[Any]:

return swh_storage		return swh_storage


def get_snapshot(storage, origin: str, revision_id: str) -> Optional[str]:		def get_snapshot(storage, origin: str, revision_id: str) -> Optional[str]:
"""Retrieve the snapshot targeting the revision_id for the given origin.		"""Retrieve the snapshot targeting the revision_id for the given origin.

"""		"""
all_visits = storage.origin_visit_get(origin)		next_page_token_visit = None
for visit in all_visits:		while True:
if not visit["snapshot"]:		visit_page = storage.origin_visit_get(
		origin, order=ListOrder.DESC, page_token=next_page_token_visit
		)
		next_page_token_visit = visit_page.next_page_token
		for visit in visit_page.result:
		next_page_token_visit_status = None
		while True:
		visit_status_page = storage.origin_visit_status_get(
		origin,
		visit,
		order=ListOrder.DESC,
		page_token=next_page_token_visit_status,
		)
		next_page_token_visit_status = visit_status_page.next_page_token
		for visit_status in visit_status_page.results:
		snapshot_id = visit_status.snapshot
		if snapshot_id is None:
continue		continue
detail_snapshot = storage.snapshot_get(visit["snapshot"])
		detail_snapshot = storage.snapshot_get(snapshot_id)
if not detail_snapshot:		if not detail_snapshot:
continue		continue
for branch_name, branch in detail_snapshot["branches"].items():		for branch_name, branch in detail_snapshot.branches.items():
if branch["target_type"] == "revision":		if branch.target_type == TargetType.REVISION:
revision = branch["target"]		revision = branch.target
if hash_to_hex(revision) == revision_id:		if hash_to_hex(revision) == revision_id:
# Found the snapshot		# Found the snapshot
return hash_to_hex(visit["snapshot"])		return hash_to_hex(snapshot_id)
		if next_page_token_visit_status is None:
		break

		if next_page_token_visit is None:
		break

return None		return None


def migrate_deposit_swhid_context_not_null(apps, schema_editor):		def migrate_deposit_swhid_context_not_null(apps, schema_editor):
"""Migrate deposit SWHIDs to the new format.		"""Migrate deposit SWHIDs to the new format.

Migrate deposit SWHIDs to the new format. Only deposit with status done and		Migrate deposit SWHIDs to the new format. Only deposit with status done and
swh_id_context not null are concerned.		swh_id_context not null are concerned.
▲ Show 20 Lines • Show All 267 Lines • Show Last 20 Lines