Skip to content

models

This module contains the metadata (and other) models that are used in the kiara_plugin.onboarding package.

Those models are convenience wrappers that make it easier for kiara to find, create, manage and version metadata -- but also other type of models -- that is attached to data, as well as kiara modules.

Metadata models must be a sub-class of [kiara.metadata.MetadataModel][]. Other models usually sub-class a pydantic BaseModel or implement custom base classes.

Classes

OnboardDataModel

Bases: KiaraModel

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
20
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
class OnboardDataModel(KiaraModel):

    _kiara_model_id: str = None  # type: ignore

    @classmethod
    def get_config_fields(cls) -> List[str]:
        return sorted(cls.__fields__.keys())

    @classmethod
    @abstractmethod
    def accepts_uri(cls, uri: str) -> Tuple[bool, str]:
        pass

    @classmethod
    def accepts_bundle_uri(cls, uri: str) -> Tuple[bool, str]:
        return cls.accepts_uri(uri)

    @abstractmethod
    def retrieve(
        self, uri: str, file_name: Union[None, str], attach_metadata: bool
    ) -> KiaraFile:
        pass

    def retrieve_bundle(
        self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
    ) -> KiaraFileBundle:
        raise NotImplementedError()

Functions

get_config_fields() -> List[str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
24
25
26
@classmethod
def get_config_fields(cls) -> List[str]:
    return sorted(cls.__fields__.keys())
accepts_uri(uri: str) -> Tuple[bool, str] classmethod abstractmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
28
29
30
31
@classmethod
@abstractmethod
def accepts_uri(cls, uri: str) -> Tuple[bool, str]:
    pass
accepts_bundle_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
33
34
35
@classmethod
def accepts_bundle_uri(cls, uri: str) -> Tuple[bool, str]:
    return cls.accepts_uri(uri)
retrieve(uri: str, file_name: Union[None, str], attach_metadata: bool) -> KiaraFile abstractmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
37
38
39
40
41
@abstractmethod
def retrieve(
    self, uri: str, file_name: Union[None, str], attach_metadata: bool
) -> KiaraFile:
    pass
retrieve_bundle(uri: str, import_config: FolderImportConfig, attach_metadata: bool) -> KiaraFileBundle
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
43
44
45
46
def retrieve_bundle(
    self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
) -> KiaraFileBundle:
    raise NotImplementedError()

FileFromLocalModel

Bases: OnboardDataModel

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
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
class FileFromLocalModel(OnboardDataModel):

    _kiara_model_id: str = "onboarding.file.from.local_file"

    @classmethod
    def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

        if os.path.isfile(os.path.abspath(uri)):
            return True, "local file exists and is file"
        else:
            return False, "local file does not exist or is not a file"

    @classmethod
    def accepts_bundle_uri(cls, uri: str) -> Tuple[bool, str]:

        if os.path.isdir(os.path.abspath(uri)):
            return True, "local folder exists and is folder"
        else:
            return False, "local folder does not exist or is not a folder"

    def retrieve(
        self, uri: str, file_name: Union[None, str], attach_metadata: bool
    ) -> KiaraFile:

        if not os.path.exists(os.path.abspath(uri)):
            raise KiaraException(
                f"Can't create file from path '{uri}': path does not exist."
            )
        if not os.path.isfile(os.path.abspath(uri)):
            raise KiaraException(
                f"Can't create file from path '{uri}': path is not a file."
            )

        return KiaraFile.load_file(uri)

    def retrieve_bundle(
        self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
    ) -> KiaraFileBundle:

        if not os.path.exists(os.path.abspath(uri)):
            raise KiaraException(
                f"Can't create file from path '{uri}': path does not exist."
            )
        if not os.path.isdir(os.path.abspath(uri)):
            raise KiaraException(
                f"Can't create file from path '{uri}': path is not a directory."
            )

        return KiaraFileBundle.import_folder(source=uri, import_config=import_config)

Functions

accepts_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
53
54
55
56
57
58
59
@classmethod
def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

    if os.path.isfile(os.path.abspath(uri)):
        return True, "local file exists and is file"
    else:
        return False, "local file does not exist or is not a file"
accepts_bundle_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
61
62
63
64
65
66
67
@classmethod
def accepts_bundle_uri(cls, uri: str) -> Tuple[bool, str]:

    if os.path.isdir(os.path.abspath(uri)):
        return True, "local folder exists and is folder"
    else:
        return False, "local folder does not exist or is not a folder"
retrieve(uri: str, file_name: Union[None, str], attach_metadata: bool) -> KiaraFile
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def retrieve(
    self, uri: str, file_name: Union[None, str], attach_metadata: bool
) -> KiaraFile:

    if not os.path.exists(os.path.abspath(uri)):
        raise KiaraException(
            f"Can't create file from path '{uri}': path does not exist."
        )
    if not os.path.isfile(os.path.abspath(uri)):
        raise KiaraException(
            f"Can't create file from path '{uri}': path is not a file."
        )

    return KiaraFile.load_file(uri)
retrieve_bundle(uri: str, import_config: FolderImportConfig, attach_metadata: bool) -> KiaraFileBundle
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def retrieve_bundle(
    self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
) -> KiaraFileBundle:

    if not os.path.exists(os.path.abspath(uri)):
        raise KiaraException(
            f"Can't create file from path '{uri}': path does not exist."
        )
    if not os.path.isdir(os.path.abspath(uri)):
        raise KiaraException(
            f"Can't create file from path '{uri}': path is not a directory."
        )

    return KiaraFileBundle.import_folder(source=uri, import_config=import_config)

FileFromRemoteModel

Bases: OnboardDataModel

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
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
class FileFromRemoteModel(OnboardDataModel):

    _kiara_model_id: str = "onboarding.file.from.url"

    @classmethod
    def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

        accepted_protocols = ["http", "https"]
        for protocol in accepted_protocols:
            if uri.startswith(f"{protocol}://"):
                return True, "url is valid (starts with http or https)"

        return False, "url is not valid (does not start with http or https)"

    def retrieve(
        self, uri: str, file_name: Union[None, str], attach_metadata: bool
    ) -> KiaraFile:
        from kiara_plugin.onboarding.utils.download import download_file

        result_file: KiaraFile = download_file(  # type: ignore
            url=uri, file_name=file_name, attach_metadata=attach_metadata
        )
        return result_file

    def retrieve_bundle(
        self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
    ) -> KiaraFileBundle:
        from kiara_plugin.onboarding.utils.download import download_file_bundle

        result_bundle = download_file_bundle(
            url=uri, import_config=import_config, attach_metadata=attach_metadata
        )
        return result_bundle

Functions

accepts_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
104
105
106
107
108
109
110
111
112
@classmethod
def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

    accepted_protocols = ["http", "https"]
    for protocol in accepted_protocols:
        if uri.startswith(f"{protocol}://"):
            return True, "url is valid (starts with http or https)"

    return False, "url is not valid (does not start with http or https)"
retrieve(uri: str, file_name: Union[None, str], attach_metadata: bool) -> KiaraFile
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
114
115
116
117
118
119
120
121
122
def retrieve(
    self, uri: str, file_name: Union[None, str], attach_metadata: bool
) -> KiaraFile:
    from kiara_plugin.onboarding.utils.download import download_file

    result_file: KiaraFile = download_file(  # type: ignore
        url=uri, file_name=file_name, attach_metadata=attach_metadata
    )
    return result_file
retrieve_bundle(uri: str, import_config: FolderImportConfig, attach_metadata: bool) -> KiaraFileBundle
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
124
125
126
127
128
129
130
131
132
def retrieve_bundle(
    self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
) -> KiaraFileBundle:
    from kiara_plugin.onboarding.utils.download import download_file_bundle

    result_bundle = download_file_bundle(
        url=uri, import_config=import_config, attach_metadata=attach_metadata
    )
    return result_bundle

FileFromZenodoModel

Bases: OnboardDataModel

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
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
232
233
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
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
class FileFromZenodoModel(OnboardDataModel):

    _kiara_model_id: str = "onboarding.file.from.zenodo"

    @classmethod
    def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

        if uri.startswith("zenodo:"):
            return True, "url is valid (follows format 'zenodo:<doi>')"

        elif "/zenodo." in uri:
            return True, "url is valid (contains '/zenodo.')"

        return False, "url is not valid (does not follow format 'zenodo:<doi>')"

    def retrieve(
        self, uri: str, file_name: Union[None, str], attach_metadata: bool
    ) -> KiaraFile:

        import pyzenodo3

        from kiara_plugin.onboarding.utils.download import download_file

        if uri.startswith("zenodo:"):
            doi = uri[len("zenodo:") :]
        elif "/zenodo." in uri:
            doi = uri

        tokens = doi.split("/zenodo.")
        if len(tokens) != 2:
            raise KiaraException(
                msg=f"Can't parse Zenodo DOI from URI for single file download: {doi}"
            )

        path_components = tokens[1].split("/", maxsplit=1)
        if len(path_components) != 2:
            raise KiaraException(
                msg=f"Can't parse Zenodo DOI from URI for single file download: {doi}"
            )

        file_path = path_components[1]
        _doi = f"{tokens[0]}/zenodo.{path_components[0]}"

        zen = pyzenodo3.Zenodo()
        record = zen.find_record_by_doi(_doi)

        match = None
        for _available_file in record.data["files"]:
            if file_path == _available_file["key"]:
                match = _available_file
                break

        if not match:
            msg = "Available files:\n"
            for key in record.data["files"]:
                msg += f"  - {key['key']}\n"
            raise KiaraException(
                msg=f"Can't find file '{file_path}' in Zenodo record. {msg}"
            )

        url = match["links"]["self"]
        checksum = match["checksum"][4:]

        file_name = file_path.split("/")[-1]

        file_model: KiaraFile
        file_model, md5_digest = download_file(  # type: ignore
            url=url,
            target=None,
            file_name=file_name,
            attach_metadata=attach_metadata,
            return_md5_hash=True,
        )

        if checksum != md5_digest:
            raise KiaraException(
                msg=f"Can't download file '{file_name}', invalid checksum: {checksum} != {md5_digest}"
            )

        if attach_metadata:
            file_model.metadata["zenodo_record_data"] = record.data

        return file_model

    def retrieve_bundle(
        self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
    ) -> KiaraFileBundle:

        import shutil

        import pyzenodo3

        from kiara_plugin.onboarding.utils.download import download_file

        if uri.startswith("zenodo:"):
            doi = uri[len("zenodo:") :]
        elif "/zenodo." in uri:
            doi = uri

        tokens = doi.split("/zenodo.")
        if len(tokens) != 2:
            raise KiaraException(
                msg=f"Can't parse Zenodo DOI from URI for single file download: {doi}"
            )

        path_components = tokens[1].split("/", maxsplit=1)

        if len(path_components) == 2:
            zid = path_components[0]
            file_path = path_components[1]
        else:
            zid = path_components[0]
            file_path = None

        _doi = f"{tokens[0]}/zenodo.{zid}"

        if not file_path:

            zen = pyzenodo3.Zenodo()

            record = zen.find_record_by_doi(_doi)

            path = KiaraFileBundle.create_tmp_dir()
            shutil.rmtree(path, ignore_errors=True)
            path.mkdir()

            for file_data in record.data["files"]:
                url = file_data["links"]["self"]
                file_name = file_data["key"]
                checksum = file_data["checksum"][4:]

                target = os.path.join(path, file_name)
                file_model: KiaraFile
                file_model, md5_digest = download_file(  # type: ignore
                    url=url,
                    target=target,
                    file_name=file_name,
                    attach_metadata=attach_metadata,
                    return_md5_hash=True,
                )

                if checksum != md5_digest:
                    raise KiaraException(
                        msg=f"Can't download file '{file_name}', invalid checksum: {checksum} != {md5_digest}"
                    )

            bundle = KiaraFileBundle.import_folder(path.as_posix())
            if attach_metadata:
                bundle.metadata["zenodo_record_data"] = record.data

        else:

            zen = pyzenodo3.Zenodo()
            record = zen.find_record_by_doi(_doi)

            match = None
            for _available_file in record.data["files"]:
                if file_path == _available_file["key"]:
                    match = _available_file
                    break

            if not match:
                msg = "Available files:\n"
                for key in record.data["files"]:
                    msg += f"  - {key['key']}\n"
                raise KiaraException(
                    msg=f"Can't find file '{file_path}' in Zenodo record. {msg}"
                )

            url = match["links"]["self"]
            checksum = match["checksum"][4:]

            file_name = file_path.split("/")[-1]

            file_model, md5_digest = download_file(  # type: ignore
                url=url,
                target=None,
                file_name=file_name,
                attach_metadata=attach_metadata,
                return_md5_hash=True,
            )

            if checksum != md5_digest:
                raise KiaraException(
                    msg=f"Can't download file '{file_name}', invalid checksum: {checksum} != {md5_digest}"
                )

            bundle = KiaraFileBundle.from_archive_file(
                archive_file=file_model, import_config=import_config
            )
            if attach_metadata:
                bundle.metadata["zenodo_record_data"] = record.data

        return bundle

Functions

accepts_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
139
140
141
142
143
144
145
146
147
148
@classmethod
def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

    if uri.startswith("zenodo:"):
        return True, "url is valid (follows format 'zenodo:<doi>')"

    elif "/zenodo." in uri:
        return True, "url is valid (contains '/zenodo.')"

    return False, "url is not valid (does not follow format 'zenodo:<doi>')"
retrieve(uri: str, file_name: Union[None, str], attach_metadata: bool) -> KiaraFile
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
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
def retrieve(
    self, uri: str, file_name: Union[None, str], attach_metadata: bool
) -> KiaraFile:

    import pyzenodo3

    from kiara_plugin.onboarding.utils.download import download_file

    if uri.startswith("zenodo:"):
        doi = uri[len("zenodo:") :]
    elif "/zenodo." in uri:
        doi = uri

    tokens = doi.split("/zenodo.")
    if len(tokens) != 2:
        raise KiaraException(
            msg=f"Can't parse Zenodo DOI from URI for single file download: {doi}"
        )

    path_components = tokens[1].split("/", maxsplit=1)
    if len(path_components) != 2:
        raise KiaraException(
            msg=f"Can't parse Zenodo DOI from URI for single file download: {doi}"
        )

    file_path = path_components[1]
    _doi = f"{tokens[0]}/zenodo.{path_components[0]}"

    zen = pyzenodo3.Zenodo()
    record = zen.find_record_by_doi(_doi)

    match = None
    for _available_file in record.data["files"]:
        if file_path == _available_file["key"]:
            match = _available_file
            break

    if not match:
        msg = "Available files:\n"
        for key in record.data["files"]:
            msg += f"  - {key['key']}\n"
        raise KiaraException(
            msg=f"Can't find file '{file_path}' in Zenodo record. {msg}"
        )

    url = match["links"]["self"]
    checksum = match["checksum"][4:]

    file_name = file_path.split("/")[-1]

    file_model: KiaraFile
    file_model, md5_digest = download_file(  # type: ignore
        url=url,
        target=None,
        file_name=file_name,
        attach_metadata=attach_metadata,
        return_md5_hash=True,
    )

    if checksum != md5_digest:
        raise KiaraException(
            msg=f"Can't download file '{file_name}', invalid checksum: {checksum} != {md5_digest}"
        )

    if attach_metadata:
        file_model.metadata["zenodo_record_data"] = record.data

    return file_model
retrieve_bundle(uri: str, import_config: FolderImportConfig, attach_metadata: bool) -> KiaraFileBundle
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def retrieve_bundle(
    self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
) -> KiaraFileBundle:

    import shutil

    import pyzenodo3

    from kiara_plugin.onboarding.utils.download import download_file

    if uri.startswith("zenodo:"):
        doi = uri[len("zenodo:") :]
    elif "/zenodo." in uri:
        doi = uri

    tokens = doi.split("/zenodo.")
    if len(tokens) != 2:
        raise KiaraException(
            msg=f"Can't parse Zenodo DOI from URI for single file download: {doi}"
        )

    path_components = tokens[1].split("/", maxsplit=1)

    if len(path_components) == 2:
        zid = path_components[0]
        file_path = path_components[1]
    else:
        zid = path_components[0]
        file_path = None

    _doi = f"{tokens[0]}/zenodo.{zid}"

    if not file_path:

        zen = pyzenodo3.Zenodo()

        record = zen.find_record_by_doi(_doi)

        path = KiaraFileBundle.create_tmp_dir()
        shutil.rmtree(path, ignore_errors=True)
        path.mkdir()

        for file_data in record.data["files"]:
            url = file_data["links"]["self"]
            file_name = file_data["key"]
            checksum = file_data["checksum"][4:]

            target = os.path.join(path, file_name)
            file_model: KiaraFile
            file_model, md5_digest = download_file(  # type: ignore
                url=url,
                target=target,
                file_name=file_name,
                attach_metadata=attach_metadata,
                return_md5_hash=True,
            )

            if checksum != md5_digest:
                raise KiaraException(
                    msg=f"Can't download file '{file_name}', invalid checksum: {checksum} != {md5_digest}"
                )

        bundle = KiaraFileBundle.import_folder(path.as_posix())
        if attach_metadata:
            bundle.metadata["zenodo_record_data"] = record.data

    else:

        zen = pyzenodo3.Zenodo()
        record = zen.find_record_by_doi(_doi)

        match = None
        for _available_file in record.data["files"]:
            if file_path == _available_file["key"]:
                match = _available_file
                break

        if not match:
            msg = "Available files:\n"
            for key in record.data["files"]:
                msg += f"  - {key['key']}\n"
            raise KiaraException(
                msg=f"Can't find file '{file_path}' in Zenodo record. {msg}"
            )

        url = match["links"]["self"]
        checksum = match["checksum"][4:]

        file_name = file_path.split("/")[-1]

        file_model, md5_digest = download_file(  # type: ignore
            url=url,
            target=None,
            file_name=file_name,
            attach_metadata=attach_metadata,
            return_md5_hash=True,
        )

        if checksum != md5_digest:
            raise KiaraException(
                msg=f"Can't download file '{file_name}', invalid checksum: {checksum} != {md5_digest}"
            )

        bundle = KiaraFileBundle.from_archive_file(
            archive_file=file_model, import_config=import_config
        )
        if attach_metadata:
            bundle.metadata["zenodo_record_data"] = record.data

    return bundle

FileFromZoteroModel

Bases: OnboardDataModel

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
331
332
333
334
335
336
337
338
339
340
class FileFromZoteroModel(OnboardDataModel):

    _kiara_model_id = "onboarding.file.from.zotero"

    @classmethod
    def accepts_uri(cls, uri: str) -> Tuple[bool, str]:
        if uri.startswith("zotero:"):
            return True, "uri is a zotero uri"
        else:
            return False, "uri is not a zotero uri, must start with 'zotero:'"

Functions

accepts_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
335
336
337
338
339
340
@classmethod
def accepts_uri(cls, uri: str) -> Tuple[bool, str]:
    if uri.startswith("zotero:"):
        return True, "uri is a zotero uri"
    else:
        return False, "uri is not a zotero uri, must start with 'zotero:'"

FileFromGithubModel

Bases: OnboardDataModel

Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
class FileFromGithubModel(OnboardDataModel):

    _kiara_model_id: str = "onboarding.file.from.github"

    @classmethod
    def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

        if uri.startswith("gh:") or uri.startswith("github:"):
            return True, "uri is a github uri"

        return False, "uri is not a github uri, must start with 'gh:' or 'github:'"

    def retrieve(
        self, uri: str, file_name: Union[None, str], attach_metadata: bool
    ) -> KiaraFile:
        from kiara_plugin.onboarding.utils.download import download_file

        tokens = uri.split(":")[1].split("/", maxsplit=3)
        if len(tokens) != 4:
            raise KiaraException(
                msg=f"Can't parse github uri '{uri}' for single file download. Required format: 'gh:<user>/<repo>/<branch_or_tag>/<path>'"
            )

        url = f"https://raw.githubusercontent.com/{tokens[0]}/{tokens[1]}/{tokens[2]}/{tokens[3]}"

        result_file: KiaraFile = download_file(  # type: ignore
            url=url, attach_metadata=attach_metadata
        )
        return result_file

    def retrieve_bundle(
        self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
    ) -> KiaraFileBundle:

        from kiara_plugin.onboarding.utils.download import download_file

        tokens = uri.split(":")[1].split("/", maxsplit=3)
        if len(tokens) == 3:
            sub_path = None
        elif len(tokens) != 4:
            raise KiaraException(
                msg=f"Can't parse github uri '{uri}' for single file download. Required format: 'gh:<user>/<repo>/<branch_or_tag>/<path>'"
            )
        else:
            sub_path = tokens[3]

        url = f"https://github.com/{tokens[0]}/{tokens[1]}/archive/refs/heads/{tokens[2]}.zip"
        file_name = f"{tokens[1]}-{tokens[2]}.zip"

        archive_zip: KiaraFile
        archive_zip = download_file(  # type: ignore
            url=url,
            attach_metadata=attach_metadata,
            file_name=file_name,
            return_md5_hash=False,
        )

        base_sub_path = f"{tokens[1]}-{tokens[2]}"

        if sub_path:
            if import_config.sub_path:
                new_sub_path = "/".join([base_sub_path, sub_path, import_config.sub_path])  # type: ignore
            else:
                new_sub_path = "/".join([base_sub_path, sub_path])
        elif import_config.sub_path:
            new_sub_path = "/".join([base_sub_path, import_config.sub_path])
        else:
            new_sub_path = base_sub_path

        import_config_new = import_config.copy(update={"sub_path": new_sub_path})

        result_bundle = KiaraFileBundle.from_archive_file(
            archive_file=archive_zip, import_config=import_config_new
        )

        return result_bundle

Functions

accepts_uri(uri: str) -> Tuple[bool, str] classmethod
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
347
348
349
350
351
352
353
@classmethod
def accepts_uri(cls, uri: str) -> Tuple[bool, str]:

    if uri.startswith("gh:") or uri.startswith("github:"):
        return True, "uri is a github uri"

    return False, "uri is not a github uri, must start with 'gh:' or 'github:'"
retrieve(uri: str, file_name: Union[None, str], attach_metadata: bool) -> KiaraFile
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def retrieve(
    self, uri: str, file_name: Union[None, str], attach_metadata: bool
) -> KiaraFile:
    from kiara_plugin.onboarding.utils.download import download_file

    tokens = uri.split(":")[1].split("/", maxsplit=3)
    if len(tokens) != 4:
        raise KiaraException(
            msg=f"Can't parse github uri '{uri}' for single file download. Required format: 'gh:<user>/<repo>/<branch_or_tag>/<path>'"
        )

    url = f"https://raw.githubusercontent.com/{tokens[0]}/{tokens[1]}/{tokens[2]}/{tokens[3]}"

    result_file: KiaraFile = download_file(  # type: ignore
        url=url, attach_metadata=attach_metadata
    )
    return result_file
retrieve_bundle(uri: str, import_config: FolderImportConfig, attach_metadata: bool) -> KiaraFileBundle
Source code in /opt/hostedtoolcache/Python/3.11.4/x64/lib/python3.11/site-packages/kiara_plugin/onboarding/models/__init__.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
def retrieve_bundle(
    self, uri: str, import_config: FolderImportConfig, attach_metadata: bool
) -> KiaraFileBundle:

    from kiara_plugin.onboarding.utils.download import download_file

    tokens = uri.split(":")[1].split("/", maxsplit=3)
    if len(tokens) == 3:
        sub_path = None
    elif len(tokens) != 4:
        raise KiaraException(
            msg=f"Can't parse github uri '{uri}' for single file download. Required format: 'gh:<user>/<repo>/<branch_or_tag>/<path>'"
        )
    else:
        sub_path = tokens[3]

    url = f"https://github.com/{tokens[0]}/{tokens[1]}/archive/refs/heads/{tokens[2]}.zip"
    file_name = f"{tokens[1]}-{tokens[2]}.zip"

    archive_zip: KiaraFile
    archive_zip = download_file(  # type: ignore
        url=url,
        attach_metadata=attach_metadata,
        file_name=file_name,
        return_md5_hash=False,
    )

    base_sub_path = f"{tokens[1]}-{tokens[2]}"

    if sub_path:
        if import_config.sub_path:
            new_sub_path = "/".join([base_sub_path, sub_path, import_config.sub_path])  # type: ignore
        else:
            new_sub_path = "/".join([base_sub_path, sub_path])
    elif import_config.sub_path:
        new_sub_path = "/".join([base_sub_path, import_config.sub_path])
    else:
        new_sub_path = base_sub_path

    import_config_new = import_config.copy(update={"sub_path": new_sub_path})

    result_bundle = KiaraFileBundle.from_archive_file(
        archive_file=archive_zip, import_config=import_config_new
    )

    return result_bundle