FileHelpers

psengine.helpers.helpers.FileHelpers

Helpers for file related functions.

read_csv staticmethod

read_csv(
    csv_file: Union[str, Path],
    as_dict: bool = False,
    single_column: bool = False,
) -> list

Reads all rows from a CSV.

It is the client's responsibility to ensure column headers are handled appropriately.

Using as_dict will read the CSV with csv.DictReader, which treats the first row as column headers. For example:

Name,ID,Level
Patrick,321,4
Ernest,123,8

as_dict=True will return:

[
    {'Name': 'Patrick', 'ID': '321', 'Level': '4'},
    {'Name': 'Ernest', 'ID': '123', 'Level': '8'}
]

single_column=True will return:

['Name', 'Patrick', 'Ernest']

as_dict=False and single_column=False will return:

[['Name', 'ID', 'Level'], ['Patrick', '321', '4'], ['Ernest', '123', '8']]

PARAMETER	DESCRIPTION
csv_file	Path to CSV file. TYPE: Union[str, Path]
as_dict	Return rows as dictionaries keyed by header. TYPE: bool DEFAULT: False
single_column	Return only first column values as strings. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
ValueError	If both as_dict and single_column are True.
ReadFileError	If file is not found or has restricted access.

RETURNS	DESCRIPTION
list	List of rows from the CSV file.

Source code in psengine/helpers/helpers.py

@staticmethod
def read_csv(
    csv_file: Annotated[Union[str, Path], Doc('Path to CSV file.')],
    as_dict: Annotated[bool, Doc('Return rows as dictionaries keyed by header.')] = False,
    single_column: Annotated[bool, Doc('Return only first column values as strings.')] = False,
) -> Annotated[list, Doc('List of rows from the CSV file.')]:
    """Reads all rows from a CSV.

    It is the client's responsibility to ensure column headers are handled appropriately.

    Using `as_dict` will read the CSV with `csv.DictReader`, which treats the first row
    as column headers. For example:

    ```csv
    Name,ID,Level
    Patrick,321,4
    Ernest,123,8
    ```

    `as_dict=True` will return:

    ```python
    [
        {'Name': 'Patrick', 'ID': '321', 'Level': '4'},
        {'Name': 'Ernest', 'ID': '123', 'Level': '8'}
    ]
    ```

    `single_column=True` will return:

    ```python
    ['Name', 'Patrick', 'Ernest']
    ```

    `as_dict=False` and `single_column=False` will return:

    ```python
    [['Name', 'ID', 'Level'], ['Patrick', '321', '4'], ['Ernest', '123', '8']]
    ```

    Raises:
        ValueError: If both `as_dict` and `single_column` are True.
        ReadFileError: If file is not found or has restricted access.
    """
    if as_dict and single_column:
        raise ValueError('Cannot use as_dict and single_column together')

    csv_file = Path(csv_file)
    if not csv_file.is_absolute():
        LOG.debug(
            f'{csv_file} is not an absolute path. Attempting to find it in {ROOT_DIR}',
        )
        file_path = Path(ROOT_DIR) / csv_file
    else:
        file_path = csv_file
    try:
        with file_path.open() as file_obj:
            if as_dict:
                reader = csv.DictReader(file_obj)
                return list(reader)

            reader = csv.reader(file_obj)
            if single_column:
                return [row[0] for row in reader]

            return list(reader)
    except OSError as oe:
        raise ReadFileError(f'Error reading entity file: {str(oe)}') from oe

write_file staticmethod

write_file(
    to_write: str,
    output_directory: Union[str, Path],
    fname: str,
) -> Path

Write string content to a file.

Creates the specified output directory if it doesn't exist. Overwrites file if it exists.

PARAMETER	DESCRIPTION
to_write	Content to write to file. TYPE: str
output_directory	Directory to write the file into. TYPE: Union[str, Path]
fname	Name of the file to write. TYPE: str

RAISES	DESCRIPTION
ValueError	If any of the parameters are invalid.
WriteFileError	If the write operation fails.

RETURNS	DESCRIPTION
Path	Path to the file written.

Source code in psengine/helpers/helpers.py

@staticmethod
def write_file(
    to_write: Annotated[str, Doc('Content to write to file.')],
    output_directory: Annotated[Union[str, Path], Doc('Directory to write the file into.')],
    fname: Annotated[str, Doc('Name of the file to write.')],
) -> Annotated[Path, Doc('Path to the file written.')]:
    """Write string content to a file.

    Creates the specified output directory if it doesn't exist. Overwrites file if it exists.

    Raises:
        ValueError: If any of the parameters are invalid.
        WriteFileError: If the write operation fails.
    """
    LOG.info(f'Writing file: {fname}')
    output_directory = Path(output_directory)
    try:
        if not output_directory.is_absolute():
            output_directory = OSHelpers.mkdir(output_directory)
        full_path = output_directory / fname
        with full_path.open('wb') as f:
            f.write(to_write)
    except OSError as err:
        raise WriteFileError(f"Error writing file '{err.filename}': {str(err)}") from err

    LOG.info(f'File written to: {full_path.as_posix()}')
    return full_path