Function get_ftp_content_and_delete

tests/test_pipeline_files.py:58–80

(
    path: str,
    host: str,
    port: int,
    username: str,
    password: str,
    use_active_mode: bool = False,
)

def get_ftp_content_and_delete(
    path: str,
    host: str,
    port: int,
    username: str,
    password: str,
    use_active_mode: bool = False,
) -> bytes:
    with FTP() as ftp:
        ftp.connect(host, port)
        ftp.login(username, password)
        if use_active_mode:
            ftp.set_pasv(False)
        ftp_data: list[bytes] = []

        def buffer_data(data: bytes) -> None:
            ftp_data.append(data)

        ftp.retrbinary(f"RETR {path}", buffer_data)
        dirname, filename = split(path)
        ftp.cwd(dirname)
        ftp.delete(filename)
    return b"".join(ftp_data)


class DeferredFSFilesStore(FSFilesStore):