Skip to content

registries

Attributes

ARCHIVE_CONFIG_CLS = TypeVar('ARCHIVE_CONFIG_CLS', bound=ArchiveConfig) module-attribute

logger = structlog.getLogger() module-attribute

NON_ARCHIVE_DETAILS = ArchiveDetails() module-attribute

Classes

ArchiveConfig

Bases: BaseModel

Source code in kiara/registries/__init__.py 
22
23
24
25
class ArchiveConfig(BaseModel):
    class Config:
        json_loads = orjson.loads
        json_dumps = orjson_dumps

Classes

Config
Source code in kiara/registries/__init__.py 
23
24
25
class Config:
    json_loads = orjson.loads
    json_dumps = orjson_dumps
Attributes
json_loads = orjson.loads class-attribute
json_dumps = orjson_dumps class-attribute

ArchiveDetails

Bases: BaseModel

Source code in kiara/registries/__init__.py 
34
35
36
37
38
class ArchiveDetails(BaseModel):

    size: Union[int, None] = Field(
        description="The size of the stored archive.", default=None
    )

Attributes

size: Union[int, None] = Field(description='The size of the stored archive.', default=None) class-attribute

KiaraArchive

Bases: abc.ABC

Source code in kiara/registries/__init__.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class KiaraArchive(abc.ABC):

    _config_cls = ArchiveConfig  # type: ignore

    @classmethod
    @abc.abstractmethod
    def supported_item_types(cls) -> Iterable[str]:
        pass

    @classmethod
    @abc.abstractmethod
    def is_writeable(cls) -> bool:
        pass

    @abc.abstractmethod
    def register_archive(self, kiara: "Kiara"):
        pass

    @abc.abstractmethod
    def retrieve_archive_id(self) -> uuid.UUID:
        pass

    @property
    def archive_id(self) -> uuid.UUID:
        return self.retrieve_archive_id()

    @property
    def config(self) -> ArchiveConfig:
        return self._get_config()

    @abc.abstractmethod
    def _get_config(self) -> ArchiveConfig:
        pass

    def get_archive_details(self) -> ArchiveDetails:
        return NON_ARCHIVE_DETAILS

    def delete_archive(self, archive_id: Union[uuid.UUID, None] = None):

        if archive_id != self.archive_id:
            raise Exception(
                f"Not deleting archive with id '{self.archive_id}': confirmation id '{archive_id}' does not match."
            )

        logger.info(
            "deleteing.archive",
            archive_id=self.archive_id,
            item_types=self.supported_item_types(),
            archive_type=self.__class__.__name__,
        )
        self._delete_archive()

    @abc.abstractmethod
    def _delete_archive(self):
        pass

    def __hash__(self):
        return hash(self.archive_id)

    def __eq__(self, other):

        if not isinstance(other, self.__class__):
            return False

        return self.archive_id == other.archive_id

Attributes

_config_cls = ArchiveConfig class-attribute
archive_id: uuid.UUID property
config: ArchiveConfig property

Functions

supported_item_types() -> Iterable[str] abstractmethod classmethod
Source code in kiara/registries/__init__.py 
48
49
50
51
@classmethod
@abc.abstractmethod
def supported_item_types(cls) -> Iterable[str]:
    pass
is_writeable() -> bool abstractmethod classmethod
Source code in kiara/registries/__init__.py 
53
54
55
56
@classmethod
@abc.abstractmethod
def is_writeable(cls) -> bool:
    pass
register_archive(kiara: Kiara) abstractmethod
Source code in kiara/registries/__init__.py 
58
59
60
@abc.abstractmethod
def register_archive(self, kiara: "Kiara"):
    pass
retrieve_archive_id() -> uuid.UUID abstractmethod
Source code in kiara/registries/__init__.py 
62
63
64
@abc.abstractmethod
def retrieve_archive_id(self) -> uuid.UUID:
    pass
get_archive_details() -> ArchiveDetails
Source code in kiara/registries/__init__.py 
78
79
def get_archive_details(self) -> ArchiveDetails:
    return NON_ARCHIVE_DETAILS
delete_archive(archive_id: Union[uuid.UUID, None] = None)
Source code in kiara/registries/__init__.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def delete_archive(self, archive_id: Union[uuid.UUID, None] = None):

    if archive_id != self.archive_id:
        raise Exception(
            f"Not deleting archive with id '{self.archive_id}': confirmation id '{archive_id}' does not match."
        )

    logger.info(
        "deleteing.archive",
        archive_id=self.archive_id,
        item_types=self.supported_item_types(),
        archive_type=self.__class__.__name__,
    )
    self._delete_archive()

BaseArchive

Bases: KiaraArchive, Generic[ARCHIVE_CONFIG_CLS]

Source code in kiara/registries/__init__.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class BaseArchive(KiaraArchive, Generic[ARCHIVE_CONFIG_CLS]):

    _config_cls: Type[ARCHIVE_CONFIG_CLS] = ArchiveConfig  # type: ignore

    def __init__(self, archive_id: uuid.UUID, config: ARCHIVE_CONFIG_CLS):

        self._archive_id: uuid.UUID = archive_id
        self._config: ARCHIVE_CONFIG_CLS = config
        self._kiara: Union["Kiara", None] = None

    def _get_config(self) -> ARCHIVE_CONFIG_CLS:
        return self._config

    def retrieve_archive_id(self) -> uuid.UUID:
        return self._archive_id

    @property
    def kiara_context(self) -> "Kiara":
        if self._kiara is None:
            raise Exception("Archive not registered into a kiara context yet.")
        return self._kiara

    def register_archive(self, kiara: "Kiara"):
        if self._kiara is not None:
            raise Exception("Archive already registered in a context.")
        self._kiara = kiara

    def _delete_archive(self):

        logger.info(
            "ignore.archive_delete_request",
            reason="not implemented/applicable",
            archive_id=self.archive_id,
            item_types=self.supported_item_types(),
            archive_type=self.__class__.__name__,
        )

Attributes

_config_cls: Type[ARCHIVE_CONFIG_CLS] = ArchiveConfig class-attribute
kiara_context: Kiara property

Functions

retrieve_archive_id() -> uuid.UUID
Source code in kiara/registries/__init__.py 
124
125
def retrieve_archive_id(self) -> uuid.UUID:
    return self._archive_id
register_archive(kiara: Kiara)
Source code in kiara/registries/__init__.py 
133
134
135
136
def register_archive(self, kiara: "Kiara"):
    if self._kiara is not None:
        raise Exception("Archive already registered in a context.")
    self._kiara = kiara

FileSystemArchiveConfig

Bases: ArchiveConfig

Source code in kiara/registries/__init__.py 
149
150
151
152
153
class FileSystemArchiveConfig(ArchiveConfig):

    archive_path: str = Field(
        description="The path where the data for this archive is stored."
    )

Attributes

archive_path: str = Field(description='The path where the data for this archive is stored.') class-attribute

Functions