Differential D5946 Diff 21445 swh/provenance/postgresql/provenancedb_without_path.py

Changeset View

Standalone View

swh/provenance/postgresql/provenancedb_without_path.py

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

	import psycopg2			import psycopg2
	import psycopg2.extras			import psycopg2.extras

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

				from ..provenance import ProvenanceResult
	from .provenancedb_base import ProvenanceDBBase			from .provenancedb_base import ProvenanceDBBase


	class ProvenanceWithoutPathDB(ProvenanceDBBase):			class ProvenanceWithoutPathDB(ProvenanceDBBase):
	def content_find_first(			def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
	self, id: Sha1Git			sql = """
	) -> Optional[Tuple[Sha1Git, Sha1Git, datetime, bytes]]:			SELECT C.sha1 AS content,
	self.cursor.execute(			R.sha1 AS revision,
	"""
	SELECT C.sha1 AS blob,
	R.sha1 AS rev,
	R.date AS date,			R.date AS date,
				O.url AS origin,
	'\\x'::bytea as path			'\\x'::bytea as path
	FROM content AS C			FROM content AS C
	INNER JOIN content_in_revision AS CR ON (CR.content = C.id)			INNER JOIN content_in_revision AS CR ON (CR.content=C.id)
	INNER JOIN revision as R ON (CR.revision = R.id)			INNER JOIN revision as R ON (CR.revision=R.id)
				LEFT JOIN origin as O ON (R.origin=O.id)
	WHERE C.sha1=%s			WHERE C.sha1=%s
	ORDER BY date, rev ASC LIMIT 1			ORDER BY date, revision, origin ASC LIMIT 1
	""",			"""
	(id,),			self.cursor.execute(sql, (id,))
	)			row = self.cursor.fetchone()
				vlorentzUnsubmitted Done Inline Actions ditto (and this "row -> ProvenanceResult" code should probably be unified somewhere) vlorentz: ditto (and this "row -> ProvenanceResult" code should probably be unified somewhere)
				aevisoAuthorUnsubmitted Done Inline Actions Same as above, using `psycopg2.extras.DictCursor` will simplify this aeviso: Same as above, using `psycopg2.extras.DictCursor` will simplify this
	return self.cursor.fetchone()			return ProvenanceResult(**row) if row is not None else None

	def content_find_all(			def content_find_all(
	self, id: Sha1Git, limit: Optional[int] = None			self, id: Sha1Git, limit: Optional[int] = None
	) -> Generator[Tuple[Sha1Git, Sha1Git, datetime, bytes], None, None]:			) -> Generator[ProvenanceResult, None, None]:
	early_cut = f"LIMIT {limit}" if limit is not None else ""			early_cut = f"LIMIT {limit}" if limit is not None else ""
	self.cursor.execute(			sql = f"""
	f"""			(SELECT C.sha1 AS content,
	(SELECT C.sha1 AS blob,			R.sha1 AS revision,
	R.sha1 AS rev,
	R.date AS date,			R.date AS date,
				O.url AS origin,
	'\\x'::bytea as path			'\\x'::bytea as path
	FROM content AS C			FROM content AS C
	INNER JOIN content_in_revision AS CR ON (CR.content = C.id)			INNER JOIN content_in_revision AS CR ON (CR.content=C.id)
	INNER JOIN revision AS R ON (CR.revision = R.id)			INNER JOIN revision AS R ON (CR.revision=R.id)
				LEFT JOIN origin as O ON (R.origin=O.id)
	WHERE C.sha1=%s)			WHERE C.sha1=%s)
	UNION			UNION
	(SELECT C.sha1 AS content,			(SELECT C.sha1 AS content,
	R.sha1 AS revision,			R.sha1 AS revision,
	R.date AS date,			R.date AS date,
				O.url AS origin,
	'\\x'::bytea as path			'\\x'::bytea as path
	FROM content AS C			FROM content AS C
	INNER JOIN content_in_directory AS CD ON (C.id = CD.content)			INNER JOIN content_in_directory AS CD ON (C.id=CD.content)
	INNER JOIN directory_in_revision AS DR ON (CD.directory = DR.directory)			INNER JOIN directory_in_revision AS DR ON (CD.directory=DR.directory)
	INNER JOIN revision AS R ON (DR.revision = R.id)			INNER JOIN revision AS R ON (DR.revision=R.id)
				LEFT JOIN origin as O ON (R.origin=O.id)
	WHERE C.sha1=%s)			WHERE C.sha1=%s)
	ORDER BY date, rev, path {early_cut}			ORDER BY date, revision, origin {early_cut}
	""",			"""
	(id, id),			self.cursor.execute(sql, (id, id))
				vlorentzUnsubmitted Done Inline Actions ditto vlorentz: ditto
				aevisoAuthorUnsubmitted Done Inline Actions agreed aeviso: agreed
	)			yield from (ProvenanceResult(**row) for row in self.cursor.fetchall())
	yield from self.cursor.fetchall()

	def insert_relation(self, relation: str, data: Set[Tuple[Sha1Git, Sha1Git, bytes]]):			def insert_relation(self, relation: str, data: Set[Tuple[Sha1Git, Sha1Git, bytes]]):
	if data:			if data:
	assert relation in (			assert relation in (
	"content_in_revision",			"content_in_revision",
	"content_in_directory",			"content_in_directory",
	"directory_in_revision",			"directory_in_revision",
	)			)
	Show All 15 Lines