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.")