Differential D5099 Diff 18282 swh/provenance/__init__.py

Changeset View

Standalone View

swh/provenance/init.py

	from typing import TYPE_CHECKING			from typing import TYPE_CHECKING

	from .postgresql.db_utils import connect			from .postgresql.db_utils import connect

	if TYPE_CHECKING:			if TYPE_CHECKING:
	from swh.provenance.archive import ArchiveInterface			from swh.provenance.archive import ArchiveInterface
	from swh.provenance.provenance import ProvenanceInterface			from swh.provenance.provenance import ProvenanceInterface


	def get_archive(cls: str, **kwargs) -> "ArchiveInterface":			def get_archive(cls: str, **kwargs) -> "ArchiveInterface":
	if cls == "api":			if cls == "api":
	from swh.provenance.storage.archive import ArchiveStorage			from swh.provenance.storage.archive import ArchiveStorage
				from swh.storage import get_storage

	return ArchiveStorage(**kwargs["storage"])			return ArchiveStorage(get_storage(**kwargs["storage"]))
	elif cls == "direct":			elif cls == "direct":
	from swh.provenance.postgresql.archive import ArchivePostgreSQL			from swh.provenance.postgresql.archive import ArchivePostgreSQL

	conn = connect(kwargs["db"])			return ArchivePostgreSQL(connect(kwargs["db"]))
				aevisoUnsubmitted Not Done Inline Actions Not sure how `swh db` works but should we use it to handle this connection as well? aeviso: Not sure how `swh db` works but should we use it to handle this connection as well?
	return ArchivePostgreSQL(conn)
	else:			else:
	raise NotImplementedError			raise NotImplementedError


	def get_provenance(cls: str, **kwargs) -> "ProvenanceInterface":			def get_provenance(cls: str, **kwargs) -> "ProvenanceInterface":
	if cls == "local":			if cls == "local":
	conn = connect(kwargs["db"])			conn = connect(kwargs["db"])
	if kwargs.get("with_path", True):			if kwargs.get("with_path", True):
	Show All 11 Lines