Differential D5957 Diff 21409 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

	if TYPE_CHECKING:			if TYPE_CHECKING:
	from .archive import ArchiveInterface			from .archive import ArchiveInterface
	from .provenance import ProvenanceInterface, ProvenanceStorageInterface			from .provenance import ProvenanceInterface, ProvenanceStorageInterface


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

	from .storage.archive import ArchiveStorage			from .storage.archive import ArchiveStorage

	return ArchiveStorage(get_storage(**kwargs["storage"]))			return ArchiveStorage(get_storage(**kwargs["storage"]))
	elif cls == "direct":			elif cls == "direct":
				from swh.core.db import BaseDb

	from .postgresql.archive import ArchivePostgreSQL			from .postgresql.archive import ArchivePostgreSQL

	return ArchivePostgreSQL(connect(kwargs["db"]))			return ArchivePostgreSQL(BaseDb.connect(**kwargs["db"]).conn)
	else:			else:
	raise NotImplementedError			raise NotImplementedError


	def get_provenance(**kwargs) -> "ProvenanceInterface":			def get_provenance(**kwargs) -> "ProvenanceInterface":
	from .backend import ProvenanceBackend			from .backend import ProvenanceBackend

	return ProvenanceBackend(get_provenance_storage(**kwargs))			return ProvenanceBackend(get_provenance_storage(**kwargs))


	def get_provenance_storage(cls: str, **kwargs) -> "ProvenanceStorageInterface":			def get_provenance_storage(cls: str, **kwargs) -> "ProvenanceStorageInterface":
	if cls == "local":			if cls == "local":
				from swh.core.db import BaseDb

	from .postgresql.provenancedb_base import ProvenanceDBBase			from .postgresql.provenancedb_base import ProvenanceDBBase

	conn = connect(kwargs["db"])			conn = BaseDb.connect(**kwargs["db"]).conn
	flavor = ProvenanceDBBase(conn).flavor			flavor = ProvenanceDBBase(conn).flavor
	if flavor == "with-path":			if flavor == "with-path":
	from .postgresql.provenancedb_with_path import ProvenanceWithPathDB			from .postgresql.provenancedb_with_path import ProvenanceWithPathDB

	return ProvenanceWithPathDB(conn)			return ProvenanceWithPathDB(conn)
	else:			else:
	from .postgresql.provenancedb_without_path import ProvenanceWithoutPathDB			from .postgresql.provenancedb_without_path import ProvenanceWithoutPathDB

	return ProvenanceWithoutPathDB(conn)			return ProvenanceWithoutPathDB(conn)
	else:			else:
	raise NotImplementedError			raise NotImplementedError