Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/converters.py
| # Copyright (C) 2015 The Software Heritage developers | # Copyright (C) 2015 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import datetime | import datetime | ||||
| from typing import Optional, Dict | from typing import Optional, Dict | ||||
| from swh.core.utils import decode_with_escape, encode_with_unescape | from swh.core.utils import encode_with_unescape | ||||
| from swh.model import identifiers | from swh.model import identifiers | ||||
| from swh.model.hashutil import MultiHash | from swh.model.hashutil import MultiHash | ||||
| DEFAULT_AUTHOR = { | DEFAULT_AUTHOR = { | ||||
| "fullname": None, | "fullname": None, | ||||
| "name": None, | "name": None, | ||||
| "email": None, | "email": None, | ||||
| Show All 40 Lines | if (fullname, name, email) == (None, None, None): | ||||
| return None | return None | ||||
| return { | return { | ||||
| "fullname": fullname, | "fullname": fullname, | ||||
| "name": name, | "name": name, | ||||
| "email": email, | "email": email, | ||||
| } | } | ||||
| def git_headers_to_db(git_headers): | |||||
| """Convert git headers to their database representation. | |||||
| We convert the bytes to unicode by decoding them into utf-8 and replacing | |||||
| invalid utf-8 sequences with backslash escapes. | |||||
| """ | |||||
| ret = [] | |||||
| for key, values in git_headers: | |||||
| if isinstance(values, list): | |||||
| ret.append([key, [decode_with_escape(value) for value in values]]) | |||||
| else: | |||||
| ret.append([key, decode_with_escape(values)]) | |||||
| return ret | |||||
| def db_to_git_headers(db_git_headers): | def db_to_git_headers(db_git_headers): | ||||
| ret = [] | ret = [] | ||||
| for key, values in db_git_headers: | for key, value in db_git_headers: | ||||
| if isinstance(values, list): | ret.append([key.encode("utf-8"), encode_with_unescape(value)]) | ||||
| ret.append([key, [encode_with_unescape(value) for value in values]]) | |||||
| else: | |||||
| ret.append([key, encode_with_unescape(values)]) | |||||
| return ret | return ret | ||||
| def db_to_date(date, offset, neg_utc_offset): | def db_to_date(date, offset, neg_utc_offset): | ||||
| """Convert the DB representation of a date to a swh-model compatible date. | """Convert the DB representation of a date to a swh-model compatible date. | ||||
| Args: | Args: | ||||
| date (datetime.datetime): a date pulled out of the database | date (datetime.datetime): a date pulled out of the database | ||||
| ▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines | def revision_to_db(rev): | ||||
| """ | """ | ||||
| revision = rev.to_dict() | revision = rev.to_dict() | ||||
| author = author_to_db(revision["author"]) | author = author_to_db(revision["author"]) | ||||
| date = date_to_db(revision["date"]) | date = date_to_db(revision["date"]) | ||||
| committer = author_to_db(revision["committer"]) | committer = author_to_db(revision["committer"]) | ||||
| committer_date = date_to_db(revision["committer_date"]) | committer_date = date_to_db(revision["committer_date"]) | ||||
| metadata = revision["metadata"] | |||||
| if metadata and "extra_headers" in metadata: | |||||
| metadata = metadata.copy() | |||||
| extra_headers = git_headers_to_db(metadata["extra_headers"]) | |||||
| metadata["extra_headers"] = extra_headers | |||||
| return { | return { | ||||
| "id": revision["id"], | "id": revision["id"], | ||||
| "author_fullname": author["fullname"], | "author_fullname": author["fullname"], | ||||
| "author_name": author["name"], | "author_name": author["name"], | ||||
| "author_email": author["email"], | "author_email": author["email"], | ||||
| "date": date["timestamp"], | "date": date["timestamp"], | ||||
| "date_offset": date["offset"], | "date_offset": date["offset"], | ||||
| "date_neg_utc_offset": date["neg_utc_offset"], | "date_neg_utc_offset": date["neg_utc_offset"], | ||||
| "committer_fullname": committer["fullname"], | "committer_fullname": committer["fullname"], | ||||
| "committer_name": committer["name"], | "committer_name": committer["name"], | ||||
| "committer_email": committer["email"], | "committer_email": committer["email"], | ||||
| "committer_date": committer_date["timestamp"], | "committer_date": committer_date["timestamp"], | ||||
| "committer_date_offset": committer_date["offset"], | "committer_date_offset": committer_date["offset"], | ||||
| "committer_date_neg_utc_offset": committer_date["neg_utc_offset"], | "committer_date_neg_utc_offset": committer_date["neg_utc_offset"], | ||||
| "type": revision["type"], | "type": revision["type"], | ||||
| "directory": revision["directory"], | "directory": revision["directory"], | ||||
| "message": revision["message"], | "message": revision["message"], | ||||
| "metadata": metadata, | "metadata": revision["metadata"], | ||||
| "synthetic": revision["synthetic"], | "synthetic": revision["synthetic"], | ||||
| "extra_headers": revision["extra_headers"], | |||||
| "parents": [ | "parents": [ | ||||
| {"id": revision["id"], "parent_id": parent, "parent_rank": i,} | {"id": revision["id"], "parent_id": parent, "parent_rank": i,} | ||||
| for i, parent in enumerate(revision["parents"]) | for i, parent in enumerate(revision["parents"]) | ||||
| ], | ], | ||||
| } | } | ||||
| def db_to_revision(db_revision): | def db_to_revision(db_revision): | ||||
| Show All 17 Lines | committer = db_to_author( | ||||
| db_revision["committer_email"], | db_revision["committer_email"], | ||||
| ) | ) | ||||
| committer_date = db_to_date( | committer_date = db_to_date( | ||||
| db_revision["committer_date"], | db_revision["committer_date"], | ||||
| db_revision["committer_date_offset"], | db_revision["committer_date_offset"], | ||||
| db_revision["committer_date_neg_utc_offset"], | db_revision["committer_date_neg_utc_offset"], | ||||
| ) | ) | ||||
| metadata = db_revision["metadata"] | |||||
| if metadata and "extra_headers" in metadata: | |||||
| extra_headers = db_to_git_headers(metadata["extra_headers"]) | |||||
| metadata["extra_headers"] = extra_headers | |||||
| parents = [] | parents = [] | ||||
| if "parents" in db_revision: | if "parents" in db_revision: | ||||
| for parent in db_revision["parents"]: | for parent in db_revision["parents"]: | ||||
| if parent: | if parent: | ||||
| parents.append(parent) | parents.append(parent) | ||||
| metadata = db_revision["metadata"] | |||||
| extra_headers = db_revision.get("extra_headers", ()) | |||||
| if not extra_headers and metadata and "extra_headers" in metadata: | |||||
| extra_headers = db_to_git_headers(metadata.pop("extra_headers")) | |||||
| ret = { | ret = { | ||||
| "id": db_revision["id"], | "id": db_revision["id"], | ||||
| "author": author, | "author": author, | ||||
| "date": date, | "date": date, | ||||
| "committer": committer, | "committer": committer, | ||||
| "committer_date": committer_date, | "committer_date": committer_date, | ||||
| "type": db_revision["type"], | "type": db_revision["type"], | ||||
| "directory": db_revision["directory"], | "directory": db_revision["directory"], | ||||
| "message": db_revision["message"], | "message": db_revision["message"], | ||||
| "metadata": metadata, | "metadata": metadata, | ||||
| "synthetic": db_revision["synthetic"], | "synthetic": db_revision["synthetic"], | ||||
| "extra_headers": extra_headers, | |||||
| "parents": parents, | "parents": parents, | ||||
| } | } | ||||
| if "object_id" in db_revision: | if "object_id" in db_revision: | ||||
| ret["object_id"] = db_revision["object_id"] | ret["object_id"] = db_revision["object_id"] | ||||
| return ret | return ret | ||||
| ▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines | |||||