Changeset View

Standalone View

Show All 38 Lines	elif cls == "direct":

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

return ArchivePostgreSQL(BaseDb.connect(**kwargs["db"]).conn)		return ArchivePostgreSQL(BaseDb.connect(**kwargs["db"]).conn)

elif cls == "graph":		elif cls == "graph":
try:		try:
from swh.graph.client import RemoteGraphClient		from swh.graph.client import RemoteGraphClient
		from swh.storage import get_storage

from .swhgraph.archive import ArchiveGraph		from .swhgraph.archive import ArchiveGraph

graph = RemoteGraphClient(kwargs.get("url"))		graph = RemoteGraphClient(kwargs.get("url"))
return ArchiveGraph(graph, get_storage(**kwargs["storage"]))		return ArchiveGraph(graph, get_storage(**kwargs["storage"]))

except ModuleNotFoundError:		except ModuleNotFoundError:
raise EnvironmentError(		raise EnvironmentError(
"Graph configuration required but module is not installed."		"Graph configuration required but module is not installed."
)		)
		elif cls == "multiplexer":

		from .multiplexer.archive import ArchiveMultiplexed

		archives = list((get_archive(**archive) for archive in kwargs["archives"]))
		return ArchiveMultiplexed(archives)
else:		else:
raise ValueError		raise ValueError


def get_provenance(**kwargs) -> ProvenanceInterface:		def get_provenance(**kwargs) -> ProvenanceInterface:
"""Get an provenance object with arguments ``args``.		"""Get an provenance object with arguments ``args``.

Args:		Args:
▲ Show 20 Lines • Show All 54 Lines • Show Last 20 Lines