in antlir/rpm/repo_snapshot.py [0:0]
def to_sqlite(self, repo: str, db: sqlite3.Connection) -> None:
for table, columns, gen_rows in [
(
"rpm",
self._RPM_COLUMNS,
self._gen_object_rows(
repo,
self.storage_id_to_rpm,
self._RPM_COLUMNS,
lambda rpm: {
"name": rpm.name,
"version": rpm.version,
"release": rpm.release,
"epoch": rpm.epoch,
"arch": rpm.arch,
"source_rpm": rpm.source_rpm,
},
),
),
(
"repodata",
self._REPODATA_COLUMNS,
self._gen_object_rows(
repo,
self.storage_id_to_repodata,
self._REPODATA_COLUMNS,
lambda repodata: {},
),
),
(
"repomd",
self._REPOMD_COLUMNS,
[
{
"repo": repo,
"build_timestamp": self.repomd.build_timestamp,
"metadata_xml": self.repomd.xml.decode(),
}
],
),
]:
db.executemany(
'INSERT INTO {} ("{}") VALUES ({});'.format(
table, '", "'.join(columns), ", ".join(["?"] * len(columns))
),
([d[k] for k in columns] for d in gen_rows),
)