Differential D5947 Diff 21353 swh/provenance/postgresql/provenancedb_without_path.py

Changeset View

Standalone View

swh/provenance/postgresql/provenancedb_without_path.py

from typing import Generator, Optional, Set, Tuple		from typing import Generator, Optional

import psycopg2
import psycopg2.extras

from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from ..provenance import ProvenanceResult		from ..provenance import ProvenanceResult, RelationType
from .provenancedb_base import ProvenanceDBBase		from .provenancedb_base import ProvenanceDBBase


class ProvenanceWithoutPathDB(ProvenanceDBBase):		class ProvenanceWithoutPathDB(ProvenanceDBBase):
def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
self.cursor.execute(		self.cursor.execute(
"""		"""
SELECT C.sha1 AS blob,		SELECT C.sha1 AS blob,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines	) -> Generator[ProvenanceResult, None, None]:
""",		""",
(id, id),		(id, id),
)		)
for row in self.cursor.fetchall():		for row in self.cursor.fetchall():
yield ProvenanceResult(		yield ProvenanceResult(
content=row[0], revision=row[1], date=row[2], origin=row[3], path=row[4]		content=row[0], revision=row[1], date=row[2], origin=row[3], path=row[4]
)		)

def insert_relation(self, relation: str, data: Set[Tuple[Sha1Git, Sha1Git, bytes]]):		def _relation_uses_location_table(self, relation: RelationType) -> bool:
if data:		return False
assert relation in (
"content_in_revision",
"content_in_directory",
"directory_in_revision",
)
src, dst = relation.split("_in_")

psycopg2.extras.execute_values(
self.cursor,
f"""
LOCK TABLE ONLY {relation};
INSERT INTO {relation}
SELECT {src}.id, {dst}.id
FROM (VALUES %s) AS V(src, dst)
INNER JOIN {src} on ({src}.sha1=V.src)
INNER JOIN {dst} on ({dst}.sha1=V.dst)
""",
data,
)
data.clear()