Differential D5973 Diff 21583 swh/provenance/__init__.py

Changeset View

Standalone View

swh/provenance/init.py

		# Copyright (C) 2021 The Software Heritage developers
		# See the AUTHORS file at the top-level directory of this distribution
		# License: GNU General Public License version 3, or any later version
		# See top-level LICENSE file for more information

from __future__ import annotations		from __future__ import annotations

from typing import TYPE_CHECKING		from typing import TYPE_CHECKING

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


def get_archive(cls: str, **kwargs) -> ArchiveInterface:		def get_archive(cls: str, **kwargs) -> ArchiveInterface:
"""Get an archive object of class ``cls`` with arguments ``args``.		"""Get an archive object of class ``cls`` with arguments ``args``.

Args:		Args:
cls: archive's class, either 'api' or 'direct'		cls: archive's class, either 'api' or 'direct'
args: dictionary of arguments passed to the archive class constructor		args: dictionary of arguments passed to the archive class constructor

Returns:		Returns:
an instance of archive object (either using swh.storage API or direct		an instance of archive object (either using swh.storage API or direct
queries to the archive's database)		queries to the archive's database)

Raises:		Raises:
:cls:`ValueError` if passed an unknown archive class.		:cls:`ValueError` if passed an unknown archive class.
"""		"""
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":
Show All 11 Lines	def get_provenance(**kwargs) -> ProvenanceInterface:

Args:		Args:
args: dictionary of arguments to retrieve a swh.provenance.storage		args: dictionary of arguments to retrieve a swh.provenance.storage
class (see :func:`get_provenance_storage` for details)		class (see :func:`get_provenance_storage` for details)

Returns:		Returns:
an instance of provenance object		an instance of provenance object
"""		"""
from .backend import ProvenanceBackend		from .provenance import Provenance

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


def get_provenance_storage(cls: str, **kwargs) -> ProvenanceStorageInterface:		def get_provenance_storage(cls: str, **kwargs) -> ProvenanceStorageInterface:
"""Get an archive object of class ``cls`` with arguments ``args``.		"""Get an archive object of class ``cls`` with arguments ``args``.

Args:		Args:
cls: storage's class, only 'local' is currently supported		cls: storage's class, only 'local' is currently supported
args: dictionary of arguments passed to the storage class constructor		args: dictionary of arguments passed to the storage class constructor

Returns:		Returns:
an instance of storage object		an instance of storage object

Raises:		Raises:
:cls:`ValueError` if passed an unknown archive class.		:cls:`ValueError` if passed an unknown archive class.
"""		"""
if cls == "local":		if cls == "local":
from swh.core.db import BaseDb		from swh.core.db import BaseDb

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

conn = BaseDb.connect(**kwargs["db"]).conn		conn = BaseDb.connect(**kwargs["db"]).conn
if ProvenanceDBBase(conn).flavor == "with-path":		if ProvenanceDBBase(conn).flavor == "with-path":
Show All 9 Lines