Skip to content

filesystem

Attributes

Classes

ImportLocalFileModule

Bases: KiaraModule

Import a file from the local filesystem.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
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
class ImportLocalFileModule(KiaraModule):

    """Import a file from the local filesystem."""

    _module_type_name = "import.local.file"

    def create_inputs_schema(
        self,
    ) -> ValueMapSchema:

        return {"path": {"type": "string", "doc": "The local path to the file."}}

    def create_outputs_schema(
        self,
    ) -> ValueMapSchema:

        return {"file": {"type": "file", "doc": "The loaded files."}}

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:
        return ModuleCharacteristics(is_idempotent=False)

    def process(self, inputs: ValueMap, outputs: ValueMap):

        path = inputs.get_value_data("path")

        file = KiaraFile.load_file(source=path)
        outputs.set_value("file", file)

Functions

create_inputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
34
35
36
37
38
def create_inputs_schema(
    self,
) -> ValueMapSchema:

    return {"path": {"type": "string", "doc": "The local path to the file."}}
create_outputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
40
41
42
43
44
def create_outputs_schema(
    self,
) -> ValueMapSchema:

    return {"file": {"type": "file", "doc": "The loaded files."}}
process(inputs: ValueMap, outputs: ValueMap)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
49
50
51
52
53
54
def process(self, inputs: ValueMap, outputs: ValueMap):

    path = inputs.get_value_data("path")

    file = KiaraFile.load_file(source=path)
    outputs.set_value("file", file)

DeserializeFileModule

Bases: DeserializeValueModule

Deserialize data to a 'file' value instance.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
 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
class DeserializeFileModule(DeserializeValueModule):

    """Deserialize data to a 'file' value instance."""

    _module_type_name = "deserialize.file"

    @classmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
        return {"python_object": KiaraFile}

    @classmethod
    def retrieve_serialized_value_type(cls) -> str:
        return "file"

    @classmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        return "copy"

    def to__python_object(self, data: SerializedData, **config: Any):

        keys = list(data.get_keys())
        keys.remove("__file_metadata__")
        assert len(keys) == 1

        file_metadata_chunks = data.get_serialized_data("__file_metadata__")
        assert file_metadata_chunks.get_number_of_chunks() == 1
        file_metadata_json = list(file_metadata_chunks.get_chunks(as_files=False))
        assert len(file_metadata_json) == 1
        file_metadata = orjson.loads(file_metadata_json[0])

        chunks = data.get_serialized_data(keys[0])
        assert chunks.get_number_of_chunks() == 1

        files = list(chunks.get_chunks(as_files=True, symlink_ok=True))
        assert len(files) == 1

        file: str = files[0]  # type: ignore

        _file_name = file_metadata.pop("file_name")
        _file_metadata = file_metadata.pop("metadata")
        _file_metadata_schemas = file_metadata.pop("metadata_schemas")

        fm = KiaraFile.load_file(
            source=file,
            file_name=_file_name,
        )
        fm.metadata = _file_metadata
        fm.metadata_schemas = _file_metadata_schemas
        return fm

Functions

retrieve_supported_target_profiles() -> Mapping[str, Type] classmethod
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
63
64
65
@classmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
    return {"python_object": KiaraFile}
retrieve_serialized_value_type() -> str classmethod
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
67
68
69
@classmethod
def retrieve_serialized_value_type(cls) -> str:
    return "file"
retrieve_supported_serialization_profile() -> str classmethod
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
71
72
73
@classmethod
def retrieve_supported_serialization_profile(cls) -> str:
    return "copy"
to__python_object(data: SerializedData, **config: Any)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
 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
def to__python_object(self, data: SerializedData, **config: Any):

    keys = list(data.get_keys())
    keys.remove("__file_metadata__")
    assert len(keys) == 1

    file_metadata_chunks = data.get_serialized_data("__file_metadata__")
    assert file_metadata_chunks.get_number_of_chunks() == 1
    file_metadata_json = list(file_metadata_chunks.get_chunks(as_files=False))
    assert len(file_metadata_json) == 1
    file_metadata = orjson.loads(file_metadata_json[0])

    chunks = data.get_serialized_data(keys[0])
    assert chunks.get_number_of_chunks() == 1

    files = list(chunks.get_chunks(as_files=True, symlink_ok=True))
    assert len(files) == 1

    file: str = files[0]  # type: ignore

    _file_name = file_metadata.pop("file_name")
    _file_metadata = file_metadata.pop("metadata")
    _file_metadata_schemas = file_metadata.pop("metadata_schemas")

    fm = KiaraFile.load_file(
        source=file,
        file_name=_file_name,
    )
    fm.metadata = _file_metadata
    fm.metadata_schemas = _file_metadata_schemas
    return fm

ImportFileBundleConfig

Bases: KiaraModuleConfig

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
108
109
110
111
112
113
114
115
class ImportFileBundleConfig(KiaraModuleConfig):

    include_file_types: Union[None, List[str]] = Field(
        description="File types to include.", default=None
    )
    exclude_file_types: Union[None, List[str]] = Field(
        description="File types to include.", default=None
    )

Attributes

include_file_types: Union[None, List[str]] = Field(description='File types to include.', default=None) instance-attribute class-attribute
exclude_file_types: Union[None, List[str]] = Field(description='File types to include.', default=None) instance-attribute class-attribute

ImportLocalFileBundleModule

Bases: KiaraModule

Import a folder (file_bundle) from the local filesystem.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
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
class ImportLocalFileBundleModule(KiaraModule):

    """Import a folder (file_bundle) from the local filesystem."""

    _module_type_name = "import.local.file_bundle"
    _config_cls = ImportFileBundleConfig

    def create_inputs_schema(
        self,
    ) -> ValueMapSchema:

        return {
            "path": {"type": "string", "doc": "The local path of the folder to import."}
        }

    def create_outputs_schema(
        self,
    ) -> ValueMapSchema:

        return {
            "file_bundle": {"type": "file_bundle", "doc": "The imported file bundle."}
        }

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:
        return DEFAULT_NO_IDEMPOTENT_MODULE_CHARACTERISTICS

    def process(self, inputs: ValueMap, outputs: ValueMap):

        path = inputs.get_value_data("path")

        include = self.get_config_value("include_file_types")
        exclude = self.get_config_value("exclude_file_types")

        config = FolderImportConfig(include_files=include, exclude_files=exclude)

        file_bundle = KiaraFileBundle.import_folder(source=path, import_config=config)
        outputs.set_value("file_bundle", file_bundle)

Attributes

_config_cls = ImportFileBundleConfig instance-attribute class-attribute

Functions

create_inputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
125
126
127
128
129
130
131
def create_inputs_schema(
    self,
) -> ValueMapSchema:

    return {
        "path": {"type": "string", "doc": "The local path of the folder to import."}
    }
create_outputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
133
134
135
136
137
138
139
def create_outputs_schema(
    self,
) -> ValueMapSchema:

    return {
        "file_bundle": {"type": "file_bundle", "doc": "The imported file bundle."}
    }
process(inputs: ValueMap, outputs: ValueMap)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
144
145
146
147
148
149
150
151
152
153
154
def process(self, inputs: ValueMap, outputs: ValueMap):

    path = inputs.get_value_data("path")

    include = self.get_config_value("include_file_types")
    exclude = self.get_config_value("exclude_file_types")

    config = FolderImportConfig(include_files=include, exclude_files=exclude)

    file_bundle = KiaraFileBundle.import_folder(source=path, import_config=config)
    outputs.set_value("file_bundle", file_bundle)

DeserializeFileBundleModule

Bases: DeserializeValueModule

Deserialize data to a 'file' value instance.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
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
class DeserializeFileBundleModule(DeserializeValueModule):

    """Deserialize data to a 'file' value instance."""

    _module_type_name = "deserialize.file_bundle"

    @classmethod
    def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
        return {"python_object": KiaraFileBundle}

    @classmethod
    def retrieve_serialized_value_type(cls) -> str:
        return "file_bundle"

    @classmethod
    def retrieve_supported_serialization_profile(cls) -> str:
        return "copy"

    def to__python_object(self, data: SerializedData, **config: Any):

        keys = list(data.get_keys())
        keys.remove("__file_metadata__")

        file_metadata_chunks = data.get_serialized_data("__file_metadata__")
        assert file_metadata_chunks.get_number_of_chunks() == 1
        file_metadata_json = list(file_metadata_chunks.get_chunks(as_files=False))
        assert len(file_metadata_json) == 1
        metadata = orjson.loads(file_metadata_json[0])
        file_metadata = metadata["included_files"]
        bundle_name = metadata["bundle_name"]
        # bundle_import_time = metadata["import_time"]
        sum_size = metadata["size"]
        number_of_files = metadata["number_of_files"]

        included_files = {}
        for rel_path in keys:

            chunks = data.get_serialized_data(rel_path)
            assert chunks.get_number_of_chunks() == 1

            files = list(chunks.get_chunks(as_files=True, symlink_ok=True))
            assert len(files) == 1

            file: str = files[0]  # type: ignore
            file_name = file_metadata[rel_path]["file_name"]
            # import_time = file_metadata[rel_path]["import_time"]
            fm = KiaraFile.load_file(source=file, file_name=file_name)
            included_files[rel_path] = fm

        fb_metadata = metadata.pop("metadata")
        fb_metadata_schemas = metadata.pop("metadata_schemas")

        fb = KiaraFileBundle(
            included_files=included_files,
            bundle_name=bundle_name,
            # import_time=bundle_import_time,
            number_of_files=number_of_files,
            size=sum_size,
            metadata=fb_metadata,
            metadata_schemas=fb_metadata_schemas,
        )
        return fb

Functions

retrieve_supported_target_profiles() -> Mapping[str, Type] classmethod
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
163
164
165
@classmethod
def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]:
    return {"python_object": KiaraFileBundle}
retrieve_serialized_value_type() -> str classmethod
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
167
168
169
@classmethod
def retrieve_serialized_value_type(cls) -> str:
    return "file_bundle"
retrieve_supported_serialization_profile() -> str classmethod
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
171
172
173
@classmethod
def retrieve_supported_serialization_profile(cls) -> str:
    return "copy"
to__python_object(data: SerializedData, **config: Any)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
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
def to__python_object(self, data: SerializedData, **config: Any):

    keys = list(data.get_keys())
    keys.remove("__file_metadata__")

    file_metadata_chunks = data.get_serialized_data("__file_metadata__")
    assert file_metadata_chunks.get_number_of_chunks() == 1
    file_metadata_json = list(file_metadata_chunks.get_chunks(as_files=False))
    assert len(file_metadata_json) == 1
    metadata = orjson.loads(file_metadata_json[0])
    file_metadata = metadata["included_files"]
    bundle_name = metadata["bundle_name"]
    # bundle_import_time = metadata["import_time"]
    sum_size = metadata["size"]
    number_of_files = metadata["number_of_files"]

    included_files = {}
    for rel_path in keys:

        chunks = data.get_serialized_data(rel_path)
        assert chunks.get_number_of_chunks() == 1

        files = list(chunks.get_chunks(as_files=True, symlink_ok=True))
        assert len(files) == 1

        file: str = files[0]  # type: ignore
        file_name = file_metadata[rel_path]["file_name"]
        # import_time = file_metadata[rel_path]["import_time"]
        fm = KiaraFile.load_file(source=file, file_name=file_name)
        included_files[rel_path] = fm

    fb_metadata = metadata.pop("metadata")
    fb_metadata_schemas = metadata.pop("metadata_schemas")

    fb = KiaraFileBundle(
        included_files=included_files,
        bundle_name=bundle_name,
        # import_time=bundle_import_time,
        number_of_files=number_of_files,
        size=sum_size,
        metadata=fb_metadata,
        metadata_schemas=fb_metadata_schemas,
    )
    return fb

ExportFileModule

Bases: DataExportModule

Export files.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
class ExportFileModule(DataExportModule):

    """Export files."""

    _module_type_name = "export.file"

    def export__file__as__file(self, value: KiaraFile, base_path: str, name: str):

        target_path = os.path.join(base_path, value.file_name)

        shutil.copy2(value.path, target_path)

        return {"files": target_path}

Functions

export__file__as__file(value: KiaraFile, base_path: str, name: str)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
227
228
229
230
231
232
233
def export__file__as__file(self, value: KiaraFile, base_path: str, name: str):

    target_path = os.path.join(base_path, value.file_name)

    shutil.copy2(value.path, target_path)

    return {"files": target_path}

PickFileFromFileBundleModule

Bases: KiaraModule

Pick a single file from a file_bundle value.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
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
class PickFileFromFileBundleModule(KiaraModule):

    """Pick a single file from a file_bundle value."""

    _module_type_name = "file_bundle.pick.file"

    def create_inputs_schema(
        self,
    ) -> ValueMapSchema:

        return {
            "file_bundle": {"type": "file_bundle", "doc": "The file bundle."},
            "path": {"type": "string", "doc": "The relative path of the file to pick."},
        }

    def create_outputs_schema(
        self,
    ) -> ValueMapSchema:

        return {"file": {"type": "file", "doc": "The file."}}

    def process(self, inputs: ValueMap, outputs: ValueMap):

        file_bundle: KiaraFileBundle = inputs.get_value_data("file_bundle")
        path: str = inputs.get_value_data("path")

        if path not in file_bundle.included_files.keys():
            raise KiaraProcessingException(
                f"Can't pick file '{path}' from file bundle: file not available."
            )

        file: KiaraFile = file_bundle.included_files[path]

        outputs.set_value("file", file)

Functions

create_inputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
242
243
244
245
246
247
248
249
def create_inputs_schema(
    self,
) -> ValueMapSchema:

    return {
        "file_bundle": {"type": "file_bundle", "doc": "The file bundle."},
        "path": {"type": "string", "doc": "The relative path of the file to pick."},
    }
create_outputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
251
252
253
254
255
def create_outputs_schema(
    self,
) -> ValueMapSchema:

    return {"file": {"type": "file", "doc": "The file."}}
process(inputs: ValueMap, outputs: ValueMap)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
def process(self, inputs: ValueMap, outputs: ValueMap):

    file_bundle: KiaraFileBundle = inputs.get_value_data("file_bundle")
    path: str = inputs.get_value_data("path")

    if path not in file_bundle.included_files.keys():
        raise KiaraProcessingException(
            f"Can't pick file '{path}' from file bundle: file not available."
        )

    file: KiaraFile = file_bundle.included_files[path]

    outputs.set_value("file", file)

PickSubBundle

Bases: KiaraModule

Pick a sub-folder from a file_bundle, resulting in a new file_bundle.

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
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
class PickSubBundle(KiaraModule):

    """Pick a sub-folder from a file_bundle, resulting in a new file_bundle."""

    _module_type_name = "file_bundle.pick.sub_folder"

    def create_inputs_schema(
        self,
    ) -> ValueMapSchema:

        return {
            "file_bundle": {"type": "file_bundle", "doc": "The file bundle."},
            "sub_path": {
                "type": "string",
                "doc": "The relative path of the sub-folder to pick.",
            },
        }

    def create_outputs_schema(self) -> ValueMapSchema:
        return {
            "file_bundle": {
                "type": "file_bundle",
                "doc": "The picked (sub-)file_bundle.",
            }
        }

    def process(self, inputs: ValueMap, outputs: ValueMap):

        file_bundle: KiaraFileBundle = inputs.get_value_data("file_bundle")
        sub_path: str = inputs.get_value_data("sub_path")

        result = {}
        for path, file in file_bundle.included_files.items():
            if path.startswith(sub_path):
                result[path] = file

        if not result:
            raise KiaraProcessingException(
                f"Can't pick sub-folder '{sub_path}' from file bundle: no matches."
            )

        new_file_bundle: KiaraFileBundle = KiaraFileBundle.create_from_file_models(
            result, bundle_name=f"{file_bundle.bundle_name}_{sub_path}"
        )

        outputs.set_value("file_bundle", new_file_bundle)

Functions

create_inputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
278
279
280
281
282
283
284
285
286
287
288
def create_inputs_schema(
    self,
) -> ValueMapSchema:

    return {
        "file_bundle": {"type": "file_bundle", "doc": "The file bundle."},
        "sub_path": {
            "type": "string",
            "doc": "The relative path of the sub-folder to pick.",
        },
    }
create_outputs_schema() -> ValueMapSchema
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
290
291
292
293
294
295
296
def create_outputs_schema(self) -> ValueMapSchema:
    return {
        "file_bundle": {
            "type": "file_bundle",
            "doc": "The picked (sub-)file_bundle.",
        }
    }
process(inputs: ValueMap, outputs: ValueMap)
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/kiara/modules/included_core_modules/filesystem.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
def process(self, inputs: ValueMap, outputs: ValueMap):

    file_bundle: KiaraFileBundle = inputs.get_value_data("file_bundle")
    sub_path: str = inputs.get_value_data("sub_path")

    result = {}
    for path, file in file_bundle.included_files.items():
        if path.startswith(sub_path):
            result[path] = file

    if not result:
        raise KiaraProcessingException(
            f"Can't pick sub-folder '{sub_path}' from file bundle: no matches."
        )

    new_file_bundle: KiaraFileBundle = KiaraFileBundle.create_from_file_models(
        result, bundle_name=f"{file_bundle.bundle_name}_{sub_path}"
    )

    outputs.set_value("file_bundle", new_file_bundle)