Manager

psengine.fusion.fusion_mgr.FusionMgr

FusionMgr(rf_token: str = None)

Manages requests for Recorded Future Fusion files.

PARAMETER	DESCRIPTION
rf_token	Recorded Future API token. TYPE: str DEFAULT: None

Source code in psengine/fusion/fusion_mgr.py

def __init__(self, rf_token: str = None):
    """Initializes the `FusionMgr` object.

    Args:
        rf_token (str, optional): Recorded Future API token.
    """
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token else RFClient()

get_files

get_files(
    file_paths: Union[str, list[str]],
) -> list[FileGetOut]

Get one or more files.

PARAMETER	DESCRIPTION
file_paths	One or more paths to fetch TYPE: Union[str, list[str]]

Endpoint

/fusion/v3/files/

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
FusionGetFileError	If API error occurs.

RETURNS	DESCRIPTION
list[FileGetOut]	A FusionFile object with name and content of the file

Source code in psengine/fusion/fusion_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=FusionGetFileError)
def get_files(
    self, file_paths: Annotated[Union[str, list[str]], Doc('One or more paths to fetch')]
) -> Annotated[list[FileGetOut], Doc('A FusionFile object with name and content of the file')]:
    """Get one or more files.

    Endpoint:
        `/fusion/v3/files/`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        FusionGetFileError: If API error occurs.
    """
    returned_files = []
    file_paths = file_paths if isinstance(file_paths, list) else [file_paths]
    file_paths = [f'/{p}' if not p.startswith('/') else p for p in file_paths]

    for file in file_paths:
        data = self._get_file(file)
        if data:
            returned_files.append(
                FileGetOut.model_validate(
                    {'path': file, 'content': data.content, 'exists': True}
                )
            )
        else:
            returned_files.append(
                FileGetOut.model_validate({'path': file, 'content': '', 'exists': False})
            )

    return returned_files

post_file

post_file(
    file_path: Path, fusion_path: str
) -> list[FileInfoOut]

Post a file.

PARAMETER	DESCRIPTION
file_path	Path of the local file TYPE: Path
fusion_path	Path of the fusion file TYPE: str

Endpoint

/fusion/v3/files/

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
FusionPostFileError	If API error occurs or the input file cannot be read.

RETURNS	DESCRIPTION
list[FileInfoOut]	Info of the file that have been posted

Source code in psengine/fusion/fusion_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=FusionPostFileError)
def post_file(
    self,
    file_path: Annotated[Path, Doc('Path of the local file')],
    fusion_path: Annotated[str, Doc('Path of the fusion file')],
) -> Annotated[list[FileInfoOut], Doc('Info of the file that have been posted')]:
    """Post a file.

    Endpoint:
        `/fusion/v3/files/`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        FusionPostFileError: If API error occurs or the input file cannot be read.
    """
    if not file_path.exists():
        raise FusionPostFileError(f'The file {file_path} does not exist')

    data = file_path.read_bytes()

    headers = 'application/octet-stream'
    returned_data = self.rf_client.request(
        'post',
        EP_FUSION_FILES_V3 + quote(fusion_path, safe='.'),
        data=data,
        content_type_header=headers,
    ).json()

    return FileInfoOut.model_validate(returned_data)

delete_files

delete_files(
    file_paths: Union[str, list[str]],
) -> list[FileDeleteOut]

Delete one or more files.

PARAMETER	DESCRIPTION
file_paths	One or more paths to delete TYPE: Union[str, list[str]]

Endpoint

/fusion/v3/files/

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
FusionDeleteFileError	If API error occurs.

RETURNS	DESCRIPTION
list[FileDeleteOut]	A list of deleted files.

Source code in psengine/fusion/fusion_mgr.py

@debug_call
@validate_call
def delete_files(
    self, file_paths: Annotated[Union[str, list[str]], Doc('One or more paths to delete')]
) -> Annotated[list[FileDeleteOut], Doc('A list of deleted files.')]:
    """Delete one or more files.

    Endpoint:
        `/fusion/v3/files/`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        FusionDeleteFileError: If API error occurs.
    """
    returned_files = []
    file_paths = file_paths if isinstance(file_paths, list) else [file_paths]
    file_paths = [f'/{p}' if not p.startswith('/') else p for p in file_paths]

    for file in file_paths:
        data = self._delete_file(file)
        returned_files.append(
            FileDeleteOut.model_validate({'path': file, 'deleted': bool(data)})
        )

    return returned_files

head_files

head_files(
    file_paths: Union[str, list[str]],
) -> list[FileHeadOut]

Head of one or more files.

PARAMETER	DESCRIPTION
file_paths	One or more paths to check TYPE: Union[str, list[str]]

Endpoint

/fusion/v3/files/

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
FusionHeadFileError	If API error occurs.

RETURNS	DESCRIPTION
list[FileHeadOut]	List of headers info for the requested files.

Source code in psengine/fusion/fusion_mgr.py

@debug_call
@validate_call
def head_files(
    self, file_paths: Annotated[Union[str, list[str]], Doc('One or more paths to check')]
) -> Annotated[list[FileHeadOut], Doc('List of headers info for the requested files.')]:
    """Head of one or more files.

    Endpoint:
        `/fusion/v3/files/`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        FusionHeadFileError: If API error occurs.
    """
    returned_files = []
    file_paths = file_paths if isinstance(file_paths, list) else [file_paths]
    file_paths = [f'/{p}' if not p.startswith('/') else p for p in file_paths]

    for file in file_paths:
        data = self._head_file(file)
        if data:
            returned_files.append(
                FileHeadOut.model_validate({'path': file, 'exists': True, **data.headers})
            )
        else:
            returned_files.append(FileHeadOut.model_validate({'path': file, 'exists': False}))

    return returned_files

list_dir

list_dir(file_path: str) -> DirectoryListOut

Get directory, subdirectory and file information of a path.

PARAMETER	DESCRIPTION
file_path	Directory to list TYPE: str

Endpoint

/fusion/v3/files/directory

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
FusionListDirError	If API error occurs.

RETURNS	DESCRIPTION
DirectoryListOut	The tree structure.

Source code in psengine/fusion/fusion_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=FusionListDirError)
def list_dir(
    self, file_path: Annotated[str, Doc('Directory to list')]
) -> Annotated[DirectoryListOut, Doc('The tree structure.')]:
    """Get directory, subdirectory and file information of a path.

    Endpoint:
        `/fusion/v3/files/directory`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        FusionListDirError: If API error occurs.
    """
    data = self.rf_client.request('get', EP_FUSION_DIR_V3 + quote(file_path, safe='.')).json()
    return DirectoryListOut.model_validate(data)