Changeset View
Changeset View
Standalone 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 |