Apply automatic formatting

src/scicat_configuration.py

@@ -340,14 +340,14 @@ def to_dict(self) -> dict:
 
 def build_dataclass(tp: type[T], data: dict, prefixes: tuple[str, ...] = ()) -> T:
     type_hints = get_annotations(tp)
-    unused_keys = (set(data.keys()) - set(type_hints.keys()))
+    unused_keys = set(data.keys()) - set(type_hints.keys())
     if unused_keys:
         # If ``data`` contains unnecessary fields.
         unused_keys_repr = "\n\t\t- ".join(
             ".".join((*prefixes, unused_key)) for unused_key in unused_keys
         )
         # it would be nice to log the invalid keys
-        #raise ValueError(f"Invalid argument found: \n\t\t- {unused_keys_repr}")
+        # raise ValueError(f"Invalid argument found: \n\t\t- {unused_keys_repr}")
     return tp(
         **{
             key: build_dataclass(sub_tp, value, (*prefixes, key))

src/scicat_dataset.py

@@ -134,7 +134,9 @@ def convert_to_type(input_value: Any, dtype_desc: str) -> Any:
         "getitem": lambda value, recipe: value[
             recipe.field
         ],  # The only operator that takes an argument
-        "str-replace": lambda value, recipe: str(value).replace(recipe.pattern,recipe.replacement),
+        "str-replace": lambda value, recipe: str(value).replace(
+            recipe.pattern, recipe.replacement
+        ),
     }
 )
 
@@ -456,11 +458,7 @@ def _filter_by_field_type(
     return [field for field in schemas if field.field_type == field_type]
 
 
-def _render_variable_as_type(
-        value: Any,
-        variable_map: dict,
-        dtype: str
-) -> Any:
+def _render_variable_as_type(value: Any, variable_map: dict, dtype: str) -> Any:
     return convert_to_type(render_variable_value(value, variable_map), dtype)
 
 

src/scicat_kafka.py

@@ -75,10 +75,7 @@ def validate_consumer(consumer: Consumer, logger: logging.Logger) -> bool:
         return True
 
 
-def _validate_wrdn_message_type(
-        message_content: bytes,
-        logger: logging.Logger
-) -> bool:
+def _validate_wrdn_message_type(message_content: bytes, logger: logging.Logger) -> bool:
     logger.info("Message type: %s", (message_type := message_content[4:8]))
     if message_type == WRDN_FILE_IDENTIFIER:
         logger.info("WRDN message received.")
@@ -98,9 +95,7 @@ def _filter_error_encountered(
         )
         return None
     else:
-        logger.info(
-            "Message successfully deserialized."
-        )
+        logger.info("Message successfully deserialized.")
         return deserialized_message
 
 

src/scicat_metadata.py

@@ -143,12 +143,9 @@ def from_file(cls, schema_file_name: pathlib.Path) -> "MetadataSchema":
         return cls.from_dict(_load_json_schema(schema_file_name))
 
 
-def render_variable_value(
-        var_value: Any,
-        variable_registry: dict
-) -> str:
+def render_variable_value(var_value: Any, variable_registry: dict) -> str:
     # if input is not a string it converts it to string
-    output_value = var_value if isinstance(var_value,str) else json.dumps(var_value)
+    output_value = var_value if isinstance(var_value, str) else json.dumps(var_value)
 
     # If it is only one variable, then it is a simple replacement
     if (
@@ -192,15 +189,13 @@ def collect_schemas(dir_path: pathlib.Path) -> OrderedDict[str, MetadataSchema]:
         schemas[metadata_schema.id] = metadata_schema
     return schemas
 
+
 def _select_applicable_schema(
-        selector: str | dict,
-        filename: str | None = None
+    selector: str | dict, filename: str | None = None
 ) -> bool:
     if isinstance(selector, str):
         # filename:starts_with:/ess/data/coda
-        select_target_name, select_function_name, select_argument = (
-            selector.split(":")
-        )
+        select_target_name, select_function_name, select_argument = selector.split(":")
         if select_target_name in ["filename"]:
             select_target_value = filename
         else:
@@ -215,23 +210,20 @@ def _select_applicable_schema(
         output = True
         for key, conditions in selector.items():
             if key == "or":
-                output = output and any([
-                    _select_applicable_schema(item, filename)
-                    for item
-                    in conditions
-                ])
+                output = output and any(
+                    [_select_applicable_schema(item, filename) for item in conditions]
+                )
             elif key == "and":
-                output = output and all([
-                    _select_applicable_schema(item, filename)
-                    for item
-                    in conditions
-                ])
+                output = output and all(
+                    [_select_applicable_schema(item, filename) for item in conditions]
+                )
             else:
                 raise NotImplementedError("Invalid operator")
         return output
     else:
         raise Exception(f"Invalid type for schema selector {type(selector)}")
 
+
 def select_applicable_schema(
     nexus_file: pathlib.Path,
     schemas: OrderedDict[str, MetadataSchema],

src/scicat_online_ingestor.py

@@ -74,8 +74,7 @@ def _individual_message_commit(offline_ingestors, consumer, logger: logging.Logg
             else:
                 logger.error("Offline ingestor error for job id %s", job_id)
             logger.info(
-                "Removed ingestor for message with job id %s from queue",
-                job_id
+                "Removed ingestor for message with job id %s from queue", job_id
             )
             offline_ingestors.pop(job_id)
 
@@ -130,7 +129,7 @@ def main() -> None:
                 background_ingestor
                     -c configuration_file
                     --nexus-file nexus_filename
-                    --done-writing-message-file message_file_path  
+                    --done-writing-message-file message_file_path
                     # optional depending on the message_saving_options.message_output
                 """
                 cmd = [
@@ -142,8 +141,7 @@ def main() -> None:
                 ]
                 if config.ingestion.file_handling.message_to_file:
                     ingestor_directory = compose_ingestor_directory(
-                        config.ingestion.file_handling,
-                        nexus_file_path
+                        config.ingestion.file_handling, nexus_file_path
                     )
                     done_writing_message_file_path = compose_ingestor_output_file_path(
                         ingestor_directory=ingestor_directory,
@@ -156,7 +154,10 @@ def main() -> None:
                         message=message,
                         message_file_path=done_writing_message_file_path,
                     )
-                    cmd += ["--done-writing-message-file", done_writing_message_file_path]
+                    cmd += [
+                        "--done-writing-message-file",
+                        done_writing_message_file_path,
+                    ]
 
                 logger.info("Command to be run: \n\n%s\n\n", cmd)
                 if config.ingestion.dry_run:

src/scicat_path_helpers.py

@@ -6,8 +6,7 @@
 
 
 def compose_ingestor_directory(
-    fh_options: FileHandlingOptions,
-    nexus_file_path: pathlib.Path
+    fh_options: FileHandlingOptions, nexus_file_path: pathlib.Path
 ) -> pathlib.Path:
     """Select the ingestor directory based on the file path and the options."""
     directory = pathlib.Path(fh_options.ingestor_files_directory)