aria.ops.pipe_utils
1# Copyright 2022 VMware, Inc. 2# SPDX-License-Identifier: Apache-2.0 3from __future__ import annotations 4 5import json 6import logging 7from typing import Optional 8from typing import Union 9 10logger = logging.getLogger(__name__) 11 12 13def read_from_pipe(input_pipe: str) -> Optional[dict | list]: 14 logger.debug(f"Input Pipe: {input_pipe}") 15 try: 16 with open(input_pipe, "r") as input_file: 17 logger.debug(f"Opened {input_file.name}") 18 return json.load(input_file) # type: ignore[no-any-return] 19 except Exception as e: 20 logger.error("Error when reading from Input Pipe.") 21 logger.debug(e) 22 return None 23 24 25def write_to_pipe(output_pipe: str, result: Optional[dict | list]) -> None: 26 logger.debug(repr(result)) 27 logger.debug(f"Output Pipe: {output_pipe}") 28 try: 29 with open(output_pipe, "w") as output_file: 30 logger.debug(f"Opened {output_pipe}") 31 json.dump(result, output_file) 32 logger.debug(f"Closing {output_pipe}") 33 except Exception as e: 34 logger.error("Error when writing to Output Pipe.") 35 logger.debug(e) 36 logger.debug("Finished writing results to Output Pipe.")
def
read_from_pipe(input_pipe: str) -> Union[dict, list, NoneType]:
14def read_from_pipe(input_pipe: str) -> Optional[dict | list]: 15 logger.debug(f"Input Pipe: {input_pipe}") 16 try: 17 with open(input_pipe, "r") as input_file: 18 logger.debug(f"Opened {input_file.name}") 19 return json.load(input_file) # type: ignore[no-any-return] 20 except Exception as e: 21 logger.error("Error when reading from Input Pipe.") 22 logger.debug(e) 23 return None
def
write_to_pipe(output_pipe: str, result: Union[dict, list, NoneType]) -> None:
26def write_to_pipe(output_pipe: str, result: Optional[dict | list]) -> None: 27 logger.debug(repr(result)) 28 logger.debug(f"Output Pipe: {output_pipe}") 29 try: 30 with open(output_pipe, "w") as output_file: 31 logger.debug(f"Opened {output_pipe}") 32 json.dump(result, output_file) 33 logger.debug(f"Closing {output_pipe}") 34 except Exception as e: 35 logger.error("Error when writing to Output Pipe.") 36 logger.debug(e) 37 logger.debug("Finished writing results to Output Pipe.")