Skip to content

context

Classes

ContextInfo

Bases: KiaraModel

Source code in kiara/models/context.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 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
109
110
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class ContextInfo(KiaraModel):
    @classmethod
    def create_from_context_config(
        cls,
        config: KiaraContextConfig,
        context_name: Union[str, None] = None,
        runtime_config: Union[KiaraRuntimeConfig, None] = None,
    ):

        from kiara.context import Kiara

        kiara = Kiara(config=config, runtime_config=runtime_config)
        return cls.create_from_context(kiara=kiara, context_name=context_name)

    @classmethod
    def create_from_context(cls, kiara: "Kiara", context_name: Union[str, None] = None):

        errors = {}
        try:
            value_ids = list(kiara.data_registry.retrieve_all_available_value_ids())
        except Exception as e:
            errors["values"] = e
            value_ids = []

        try:
            aliases = {
                a.full_alias: a.value_id for a in kiara.alias_registry.aliases.values()
            }
        except Exception as e:
            errors["aliases"] = e
            aliases = {}

        try:
            archives_info = ArchiveGroupInfo.create_from_context(kiara=kiara)
        except Exception as e:
            errors["archives"] = e
            archives_info = ArchiveGroupInfo(item_infos={})

        comment = None
        invalid = False
        if errors:
            invalid = True
            comment = "Errors creating this context, this is most likely a bug:\n\n"
            for k, err in errors.items():
                comment = f"{comment}  - {k}: {err}"

        result = ContextInfo.construct(
            kiara_id=kiara.id,
            value_ids=value_ids,
            aliases=aliases,
            context_name=context_name,
            archives=archives_info,
            invalid=invalid,
            comment=comment,
        )
        result._kiara = kiara
        return result

    kiara_id: uuid.UUID = Field(
        description="The (globally unique) id of the kiara context."
    )
    context_name: Union[str, None] = Field(
        description="The local alias for this context."
    )
    value_ids: List[uuid.UUID] = Field(
        description="The ids of all stored values in this context."
    )
    aliases: Dict[str, uuid.UUID] = Field(
        description="All available aliases within this context (and the value ids they refer to)."
    )
    archives: ArchiveGroupInfo = Field(
        description="The archives registered in this context."
    )
    invalid: bool = Field(description="Whether this context has errors.", bool=False)
    comment: Union[str, None] = Field(
        description="(Optional) comment about this context.", default=None
    )

    _kiara: Union["Kiara", None] = PrivateAttr()

    @property
    def kiara_context(self) -> "Kiara":
        if self._kiara is None:
            raise Exception("Kiara context object not set.")
        return self._kiara

    def value_summary(self) -> Dict[str, Any]:

        sum_size = 0
        types: Dict[str, int] = {}
        internal_types: Dict[str, int] = {}
        no_of_values = len(self.value_ids)

        for value_id in self.value_ids:
            value = self.kiara_context.data_registry.get_value(value=value_id)
            sum_size = sum_size + value.value_size
            if self.kiara_context.type_registry.is_internal_type(value.data_type_name):
                if value.data_type_name not in internal_types.keys():
                    internal_types[value.data_type_name] = 1
                else:
                    internal_types[value.data_type_name] += 1
            else:
                if value.data_type_name not in types.keys():
                    types[value.data_type_name] = 1
                else:
                    types[value.data_type_name] += 1

            types.setdefault(value.data_type_name, 0)

        return {
            "size": sum_size,
            "no_values": no_of_values,
            "types": types,
            "internal_types": internal_types,
        }

    def alias_summary(self) -> Dict[str, Any]:

        sum_size = 0
        types: Dict[str, int] = {}
        internal_types: Dict[str, int] = {}
        no_of_values = len(self.value_ids)

        for alias, value_id in self.aliases.items():
            value = self.kiara_context.data_registry.get_value(value=value_id)
            sum_size = sum_size + value.value_size
            if self.kiara_context.type_registry.is_internal_type(value.data_type_name):
                if value.data_type_name not in internal_types.keys():
                    internal_types[value.data_type_name] = 1
                else:
                    internal_types[value.data_type_name] += 1
            else:
                if value.data_type_name not in types.keys():
                    types[value.data_type_name] = 1
                else:
                    types[value.data_type_name] += 1

            types.setdefault(value.data_type_name, 0)

        return {
            "size": sum_size,
            "no_values": no_of_values,
            "types": types,
            "internal_types": internal_types,
        }

    def create_renderable(self, **config: Any) -> RenderableType:

        full_details = config.get("full_details", False)
        show_value_ids = config.get("show_value_ids", False)
        show_archive_info = config.get("show_archive_info", True)

        table = Table(box=box.SIMPLE, show_header=False)

        table.add_column("Property", style="i")
        table.add_column("Value")

        if self.invalid:
            msg = "[red]This context is invalid![/red]\n"
            cmt = self.comment if self.comment else "-- no details --"
            md = Markdown(cmt)
            table.add_row(
                "",
                Group(msg, md, ""),
            )

        if self.context_name:
            table.add_row("context name", self.context_name)
        table.add_row("kiara_id", str(self.kiara_id))

        size_on_disk = humanfriendly.format_size(self.archives.combined_size)
        table.add_row("size on disk", size_on_disk)

        value_sum = self.value_summary()
        v_table = Table(box=box.SIMPLE, show_header=False)
        v_table.add_column("Property")
        v_table.add_column("Value")
        v_table.add_row("no. values", str(value_sum["no_values"]))
        v_table.add_row("combined size", format_size(value_sum["size"]))
        if full_details and show_value_ids:
            if self.value_ids:
                value_ids = sorted((str(v) for v in self.value_ids))
                v_table.add_row("value_ids", value_ids[0])
                for v_id in value_ids[1:]:
                    v_table.add_row("", v_id)
            else:
                v_table.add_row("value_ids", "")
        table.add_row("values", v_table)

        alias_sum = self.alias_summary()
        a_table = Table(box=box.SIMPLE, show_header=False)
        a_table.add_column("Property")
        a_table.add_column("Value")
        a_table.add_row("no. aliases", str(len(self.aliases)))
        a_table.add_row("combined size", format_size(alias_sum["size"]))
        if full_details:
            if self.aliases:
                aliases = sorted(self.aliases.keys())
                a_table.add_row(
                    "aliases", f"{aliases[0]} -> {self.aliases[aliases[0]]}"
                )
                for alias in aliases[1:]:
                    a_table.add_row("", f"{alias} -> {self.aliases[alias]}")
            else:
                a_table.add_row("aliases", "")
        table.add_row("aliases", a_table)

        if show_archive_info:
            table.add_row("archives", self.archives)

        return table

Attributes

kiara_id: uuid.UUID = Field(description='The (globally unique) id of the kiara context.') class-attribute
context_name: Union[str, None] = Field(description='The local alias for this context.') class-attribute
value_ids: List[uuid.UUID] = Field(description='The ids of all stored values in this context.') class-attribute
aliases: Dict[str, uuid.UUID] = Field(description='All available aliases within this context (and the value ids they refer to).') class-attribute
archives: ArchiveGroupInfo = Field(description='The archives registered in this context.') class-attribute
invalid: bool = Field(description='Whether this context has errors.', bool=False) class-attribute
comment: Union[str, None] = Field(description='(Optional) comment about this context.', default=None) class-attribute
kiara_context: Kiara property

Functions

create_from_context_config(config: KiaraContextConfig, context_name: Union[str, None] = None, runtime_config: Union[KiaraRuntimeConfig, None] = None) classmethod
Source code in kiara/models/context.py 
22
23
24
25
26
27
28
29
30
31
32
33
@classmethod
def create_from_context_config(
    cls,
    config: KiaraContextConfig,
    context_name: Union[str, None] = None,
    runtime_config: Union[KiaraRuntimeConfig, None] = None,
):

    from kiara.context import Kiara

    kiara = Kiara(config=config, runtime_config=runtime_config)
    return cls.create_from_context(kiara=kiara, context_name=context_name)
create_from_context(kiara: Kiara, context_name: Union[str, None] = None) classmethod
Source code in kiara/models/context.py 
35
36
37
38
39
40
41
42
43
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
@classmethod
def create_from_context(cls, kiara: "Kiara", context_name: Union[str, None] = None):

    errors = {}
    try:
        value_ids = list(kiara.data_registry.retrieve_all_available_value_ids())
    except Exception as e:
        errors["values"] = e
        value_ids = []

    try:
        aliases = {
            a.full_alias: a.value_id for a in kiara.alias_registry.aliases.values()
        }
    except Exception as e:
        errors["aliases"] = e
        aliases = {}

    try:
        archives_info = ArchiveGroupInfo.create_from_context(kiara=kiara)
    except Exception as e:
        errors["archives"] = e
        archives_info = ArchiveGroupInfo(item_infos={})

    comment = None
    invalid = False
    if errors:
        invalid = True
        comment = "Errors creating this context, this is most likely a bug:\n\n"
        for k, err in errors.items():
            comment = f"{comment}  - {k}: {err}"

    result = ContextInfo.construct(
        kiara_id=kiara.id,
        value_ids=value_ids,
        aliases=aliases,
        context_name=context_name,
        archives=archives_info,
        invalid=invalid,
        comment=comment,
    )
    result._kiara = kiara
    return result
value_summary() -> Dict[str, Any]
Source code in kiara/models/context.py 
107
108
109
110
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
def value_summary(self) -> Dict[str, Any]:

    sum_size = 0
    types: Dict[str, int] = {}
    internal_types: Dict[str, int] = {}
    no_of_values = len(self.value_ids)

    for value_id in self.value_ids:
        value = self.kiara_context.data_registry.get_value(value=value_id)
        sum_size = sum_size + value.value_size
        if self.kiara_context.type_registry.is_internal_type(value.data_type_name):
            if value.data_type_name not in internal_types.keys():
                internal_types[value.data_type_name] = 1
            else:
                internal_types[value.data_type_name] += 1
        else:
            if value.data_type_name not in types.keys():
                types[value.data_type_name] = 1
            else:
                types[value.data_type_name] += 1

        types.setdefault(value.data_type_name, 0)

    return {
        "size": sum_size,
        "no_values": no_of_values,
        "types": types,
        "internal_types": internal_types,
    }
alias_summary() -> Dict[str, Any]
Source code in kiara/models/context.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def alias_summary(self) -> Dict[str, Any]:

    sum_size = 0
    types: Dict[str, int] = {}
    internal_types: Dict[str, int] = {}
    no_of_values = len(self.value_ids)

    for alias, value_id in self.aliases.items():
        value = self.kiara_context.data_registry.get_value(value=value_id)
        sum_size = sum_size + value.value_size
        if self.kiara_context.type_registry.is_internal_type(value.data_type_name):
            if value.data_type_name not in internal_types.keys():
                internal_types[value.data_type_name] = 1
            else:
                internal_types[value.data_type_name] += 1
        else:
            if value.data_type_name not in types.keys():
                types[value.data_type_name] = 1
            else:
                types[value.data_type_name] += 1

        types.setdefault(value.data_type_name, 0)

    return {
        "size": sum_size,
        "no_values": no_of_values,
        "types": types,
        "internal_types": internal_types,
    }
create_renderable(**config: Any) -> RenderableType
Source code in kiara/models/context.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def create_renderable(self, **config: Any) -> RenderableType:

    full_details = config.get("full_details", False)
    show_value_ids = config.get("show_value_ids", False)
    show_archive_info = config.get("show_archive_info", True)

    table = Table(box=box.SIMPLE, show_header=False)

    table.add_column("Property", style="i")
    table.add_column("Value")

    if self.invalid:
        msg = "[red]This context is invalid![/red]\n"
        cmt = self.comment if self.comment else "-- no details --"
        md = Markdown(cmt)
        table.add_row(
            "",
            Group(msg, md, ""),
        )

    if self.context_name:
        table.add_row("context name", self.context_name)
    table.add_row("kiara_id", str(self.kiara_id))

    size_on_disk = humanfriendly.format_size(self.archives.combined_size)
    table.add_row("size on disk", size_on_disk)

    value_sum = self.value_summary()
    v_table = Table(box=box.SIMPLE, show_header=False)
    v_table.add_column("Property")
    v_table.add_column("Value")
    v_table.add_row("no. values", str(value_sum["no_values"]))
    v_table.add_row("combined size", format_size(value_sum["size"]))
    if full_details and show_value_ids:
        if self.value_ids:
            value_ids = sorted((str(v) for v in self.value_ids))
            v_table.add_row("value_ids", value_ids[0])
            for v_id in value_ids[1:]:
                v_table.add_row("", v_id)
        else:
            v_table.add_row("value_ids", "")
    table.add_row("values", v_table)

    alias_sum = self.alias_summary()
    a_table = Table(box=box.SIMPLE, show_header=False)
    a_table.add_column("Property")
    a_table.add_column("Value")
    a_table.add_row("no. aliases", str(len(self.aliases)))
    a_table.add_row("combined size", format_size(alias_sum["size"]))
    if full_details:
        if self.aliases:
            aliases = sorted(self.aliases.keys())
            a_table.add_row(
                "aliases", f"{aliases[0]} -> {self.aliases[aliases[0]]}"
            )
            for alias in aliases[1:]:
                a_table.add_row("", f"{alias} -> {self.aliases[alias]}")
        else:
            a_table.add_row("aliases", "")
    table.add_row("aliases", a_table)

    if show_archive_info:
        table.add_row("archives", self.archives)

    return table

ContextInfos

Bases: BaseModel

Source code in kiara/models/context.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
class ContextInfos(BaseModel):
    __root__: Dict[str, ContextInfo]

    @classmethod
    def create_context_infos(
        cls, contexts: Union[Mapping[str, "KiaraContextConfig"], None] = None
    ):

        if not contexts:
            kc = KiaraConfig()
            contexts = kc.context_configs

        return ContextInfos(
            __root__={
                a: ContextInfo.create_from_context_config(c, context_name=a)
                for a, c in contexts.items()
            }
        )

    def create_renderable(self, **config: Any) -> RenderableType:

        full_details = config.get("full_details", False)

        if not full_details:
            table = Table(box=box.SIMPLE, show_header=True, show_lines=False)
            table.add_column("context name", style="i")
            table.add_column("context id", style="i")
            table.add_column("size on disk")
            table.add_column("size of all values")
            table.add_column("no. values")
            table.add_column("no. aliaes")
            for context_name, context_summary in self.__root__.items():
                size_on_disk = context_summary.archives.combined_size
                value_summary = context_summary.value_summary()
                size = humanfriendly.format_size(value_summary["size"])
                no_values = str(value_summary["no_values"])
                no_aliases = str(len(context_summary.aliases))
                table.add_row(
                    context_name,
                    str(context_summary.kiara_id),
                    humanfriendly.format_size(size_on_disk),
                    size,
                    no_values,
                    no_aliases,
                )
        else:

            table = Table(box=box.MINIMAL, show_header=True, show_lines=True)
            table.add_column("context_name", style="i")
            table.add_column("details")

            for context_name, context_summary in self.__root__.items():

                table.add_row(context_name, context_summary.create_renderable(**config))

        return table

Functions

create_context_infos(contexts: Union[Mapping[str, KiaraContextConfig], None] = None) classmethod
Source code in kiara/models/context.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@classmethod
def create_context_infos(
    cls, contexts: Union[Mapping[str, "KiaraContextConfig"], None] = None
):

    if not contexts:
        kc = KiaraConfig()
        contexts = kc.context_configs

    return ContextInfos(
        __root__={
            a: ContextInfo.create_from_context_config(c, context_name=a)
            for a, c in contexts.items()
        }
    )
create_renderable(**config: Any) -> RenderableType
Source code in kiara/models/context.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def create_renderable(self, **config: Any) -> RenderableType:

    full_details = config.get("full_details", False)

    if not full_details:
        table = Table(box=box.SIMPLE, show_header=True, show_lines=False)
        table.add_column("context name", style="i")
        table.add_column("context id", style="i")
        table.add_column("size on disk")
        table.add_column("size of all values")
        table.add_column("no. values")
        table.add_column("no. aliaes")
        for context_name, context_summary in self.__root__.items():
            size_on_disk = context_summary.archives.combined_size
            value_summary = context_summary.value_summary()
            size = humanfriendly.format_size(value_summary["size"])
            no_values = str(value_summary["no_values"])
            no_aliases = str(len(context_summary.aliases))
            table.add_row(
                context_name,
                str(context_summary.kiara_id),
                humanfriendly.format_size(size_on_disk),
                size,
                no_values,
                no_aliases,
            )
    else:

        table = Table(box=box.MINIMAL, show_header=True, show_lines=True)
        table.add_column("context_name", style="i")
        table.add_column("details")

        for context_name, context_summary in self.__root__.items():

            table.add_row(context_name, context_summary.create_renderable(**config))

    return table