Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/cassandra/converters.py
| Show All 36 Lines | def revision_to_db(revision: Revision) -> Dict[str, Any]: | ||||
| db_revision["metadata"] = json.dumps( | db_revision["metadata"] = json.dumps( | ||||
| dict(db_revision["metadata"]) if db_revision["metadata"] is not None else None | dict(db_revision["metadata"]) if db_revision["metadata"] is not None else None | ||||
| ) | ) | ||||
| db_revision["extra_headers"] = extra_headers | db_revision["extra_headers"] = extra_headers | ||||
| db_revision["type"] = db_revision["type"].value | db_revision["type"] = db_revision["type"].value | ||||
| return db_revision | return db_revision | ||||
| def revision_from_db(db_revision: Row, parents: Tuple[Sha1Git]) -> Revision: | def revision_from_db(db_revision: Row, parents: Tuple[Sha1Git, ...]) -> Revision: | ||||
| revision = db_revision._asdict() # type: ignore | revision = db_revision._asdict() # type: ignore | ||||
| metadata = json.loads(revision.pop("metadata", None)) | metadata = json.loads(revision.pop("metadata", None)) | ||||
| extra_headers = revision.pop("extra_headers", ()) | extra_headers = revision.pop("extra_headers", ()) | ||||
| if not extra_headers and metadata and "extra_headers" in metadata: | if not extra_headers and metadata and "extra_headers" in metadata: | ||||
| extra_headers = metadata.pop("extra_headers") | extra_headers = metadata.pop("extra_headers") | ||||
| if extra_headers is None: | if extra_headers is None: | ||||
| extra_headers = () | extra_headers = () | ||||
| return Revision( | return Revision( | ||||
| ▲ Show 20 Lines • Show All 53 Lines • Show Last 20 Lines | |||||