Source code for ska_pst.send.file_util

# -*- coding: utf-8 -*-
#
# This file is part of the SKA PST project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.

"""Module for utility methods for moving files."""

from __future__ import annotations

import logging
import pathlib

_logger: logging.Logger = logging.getLogger(__name__)


[docs]def move_files(input_path: pathlib.Path, output_path: pathlib.Path) -> None: """ Move file(s) from input to output location. This utility method will move file(s) from the input path to the output path. If the input path is a directory, then the output path must be a directory. If the input is a file and the output path is directory then the file will be moved into the output directory. If the input is a directory then this method will attempt to move the directory but if the output directory already exists it will move the individual files to the output path and then delete the input path directory. If the input path doesn't exist then this will log a warning that it doesn't exist rather than raising an error. :param input_path: a file or directory that needs to be moved to the output path :type input_path: pathlib.Path :param output_path: a file or directory that the input file(s) will be moved to :type output_path: pathlib.Path :throws AssertionError: raised when input_path is a directory, output_path exists but is not a directory. """ _logger.debug(f"moving {input_path=} to {output_path=}") if not input_path.exists(): _logger.warning(f"Attempting to move {input_path=} to {output_path=} but input does exist.") return if input_path.is_dir(): if output_path.exists(): assert ( output_path.is_dir() ), f"Expected {output_path=} to be a directory as {input_path=} is a directory." # move individual files for f in input_path.iterdir(): move_files(f, output_path / f.name) input_path.rmdir() else: _logger.debug( f"{output_path=} doesn't exist, the move of {input_path=} will create the directory" ) input_path.rename(output_path) else: if output_path.is_dir(): output_file = output_path / input_path.name _logger.debug(f"moving {input_path=} to output {output_file}") input_path.rename(output_file) else: if output_path.exists(): _logger.warning(f"Moving {input_path=} to {output_path=} is overwriting existing file") input_path.rename(output_path)