|
20 | 20 |
|
21 | 21 | from .yaml_transformer import YamlTransformer
|
22 | 22 |
|
23 |
| -def _json_to_zs_bin(json_input_path, bin_output_path, module_name, type_name): |
| 23 | +def _yaml_to_zserio_object(yaml_input_path): |
24 | 24 | """
|
25 |
| - Converts a JSON file to a binary file using Zserio serialization. |
| 25 | + Converts a YAML file to a Zserio object. |
26 | 26 |
|
27 | 27 | Args:
|
28 |
| - json_input_path (str): Path to the input JSON file. |
29 |
| - bin_output_path (str): Path to the output binary file. |
30 |
| - module_name (str): Name of the Zserio module containing the schema. |
31 |
| - type_name (str): Name of the Zserio type to be used for serialization. |
| 28 | + yaml_input_path (str): Path to the input YAML file. |
| 29 | +
|
| 30 | + Returns: |
| 31 | + tuple: A tuple containing (zserio_object, temp_json_path, meta) |
32 | 32 |
|
33 | 33 | Raises:
|
34 |
| - ValueError: If the specified Zserio type is not found in the module. |
| 34 | + ValueError: If schema_module and schema_type are not specified in the _meta |
| 35 | + section of the YAML file. |
35 | 36 | """
|
36 |
| - module = importlib.import_module(module_name) |
37 |
| - ImportedType = getattr(module, type_name) |
| 37 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as temp_json_file: |
| 38 | + temp_json_path = temp_json_file.name |
| 39 | + |
| 40 | + meta = yaml_to_json(yaml_input_path, temp_json_path) |
| 41 | + |
| 42 | + schema_module = meta.get('schema_module') |
| 43 | + schema_type = meta.get('schema_type') |
| 44 | + init_args = meta.get('initialization_args', []) |
| 45 | + |
| 46 | + if not schema_module or not schema_type: |
| 47 | + raise ValueError("Error: schema_module and schema_type must be specified in the _meta section") |
| 48 | + |
| 49 | + module = importlib.import_module(schema_module) |
| 50 | + ImportedType = getattr(module, schema_type) |
38 | 51 | if ImportedType is None:
|
39 |
| - raise ValueError(f"Type {type_name} not found in module {module_name}") |
40 |
| - zserio_object = zserio.from_json_file(ImportedType, json_input_path) |
41 |
| - zserio.serialize_to_file(zserio_object, bin_output_path) |
| 52 | + raise ValueError(f"Type {schema_type} not found in module {schema_module}") |
| 53 | + |
| 54 | + zserio_object = zserio.from_json_file(ImportedType, temp_json_path, *init_args) |
| 55 | + return zserio_object, temp_json_path, meta |
| 56 | + |
| 57 | +def yaml_to_bin(yaml_input_path, bin_output_path): |
| 58 | + """ |
| 59 | + Converts a YAML file to a binary file using Zserio serialization. |
| 60 | +
|
| 61 | + Args: |
| 62 | + yaml_input_path (str): Path to the input YAML file. |
| 63 | + bin_output_path (str): Path to the output binary file. |
| 64 | + """ |
| 65 | + try: |
| 66 | + zserio_object, temp_json_path, _ = _yaml_to_zserio_object(yaml_input_path) |
| 67 | + zserio.serialize_to_file(zserio_object, bin_output_path) |
| 68 | + finally: |
| 69 | + os.remove(temp_json_path) |
| 70 | + |
| 71 | +def yaml_to_pyobj(yaml_input_path): |
| 72 | + """ |
| 73 | + Converts a YAML file to an in-memory Python object using Zserio deserialization. |
| 74 | +
|
| 75 | + Args: |
| 76 | + yaml_input_path (str): Path to the input YAML file. |
| 77 | +
|
| 78 | + Returns: |
| 79 | + object: The deserialized Python object. |
| 80 | + """ |
| 81 | + try: |
| 82 | + zserio_object, temp_json_path, _ = _yaml_to_zserio_object(yaml_input_path) |
| 83 | + return zserio_object |
| 84 | + finally: |
| 85 | + os.remove(temp_json_path) |
42 | 86 |
|
43 | 87 | def yaml_to_yaml(yaml_input_path, yaml_output_path=None):
|
44 | 88 | """
|
@@ -95,42 +139,6 @@ def json_to_yaml(json_input_path, yaml_output_path):
|
95 | 139 | yaml.safe_dump(data, yaml_file, default_flow_style=False, sort_keys=False)
|
96 | 140 |
|
97 | 141 |
|
98 |
| -def yaml_to_bin(yaml_input_path, bin_output_path): |
99 |
| - """ |
100 |
| - Converts a YAML file to a binary file using Zserio serialization. |
101 |
| -
|
102 |
| - Args: |
103 |
| - yaml_input_path (str): Path to the input YAML file. |
104 |
| - bin_output_path (str): Path to the output binary file. |
105 |
| -
|
106 |
| - Raises: |
107 |
| - ValueError: If schema_module and schema_type are not specified in the _meta |
108 |
| - section of the YAML file. |
109 |
| - """ |
110 |
| - with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as temp_json_file: |
111 |
| - temp_json_path = temp_json_file.name |
112 |
| - |
113 |
| - try: |
114 |
| - meta = yaml_to_json(yaml_input_path, temp_json_path) |
115 |
| - |
116 |
| - schema_module = meta.get('schema_module') |
117 |
| - schema_type = meta.get('schema_type') |
118 |
| - init_args = meta.get('initialization_args', []) |
119 |
| - |
120 |
| - if not schema_module or not schema_type: |
121 |
| - raise ValueError("Error: schema_module and schema_type must be specified in the _meta section for binary output") |
122 |
| - |
123 |
| - module = importlib.import_module(schema_module) |
124 |
| - ImportedType = getattr(module, schema_type) |
125 |
| - if ImportedType is None: |
126 |
| - raise ValueError(f"Type {schema_type} not found in module {schema_module}") |
127 |
| - |
128 |
| - zserio_object = zserio.from_json_file(ImportedType, temp_json_path, *init_args) |
129 |
| - zserio.serialize_to_file(zserio_object, bin_output_path) |
130 |
| - finally: |
131 |
| - os.remove(temp_json_path) |
132 |
| - |
133 |
| - |
134 | 142 | def bin_to_yaml(bin_input_path, yaml_output_path):
|
135 | 143 | """
|
136 | 144 | Converts a binary file to a YAML file using Zserio deserialization.
|
|
0 commit comments