def add_snapshot_db_objs()

in antlir/rpm/repo_server.py [0:0]

• 39 lines of code
• 10 McCabe index (conditional complexity)

def add_snapshot_db_objs(db):
    location_to_obj = {}
    for repo, build_timestamp, metadata_xml in db.execute(
        """
    SELECT "repo", "build_timestamp", "metadata_xml" FROM "repomd"
    """
    ).fetchall():
        set_new_key(
            location_to_obj,
            os.path.join(repo, "repodata/repomd.xml"),
            {
                "size": len(metadata_xml),
                "build_timestamp": build_timestamp,
                "content_bytes": metadata_xml.encode(),
            },
        )
    for table in ["repodata", "rpm"]:
        for (
            repo,
            path,
            build_timestamp,
            checksum,
            error,
            error_json,
            size,
            storage_id,
        ) in db.execute(
            f"""
        SELECT
            "repo", "path", "build_timestamp", "checksum", "error",
            "error_json", "size", "storage_id"
        FROM "{table}"
        """
        ).fetchall():
            obj = {
                "checksum": checksum,
                "size": size,
                "build_timestamp": build_timestamp,
            }
            # `storage_id` is populated in the DB table for `mutable_rpm`
            # errors, but we don't want to serve up those files.
            if storage_id and not error and not error_json:
                obj["storage_id"] = storage_id
            elif error and error_json:
                obj["error"] = {"error": error, **json.loads(error_json)}
            else:  # pragma: no cover
                raise AssertionError(f"{storage_id} {error} {error_json}")
            set_new_key(location_to_obj, os.path.join(repo, path), obj)
    return location_to_obj