def get_files_zip(run_id: int, filetype: _FileType):
    """Send all artifacts or sources of a run as ZIP."""
    data = current_app.config["data"]
    dao_runs = data.get_run_dao()
    dao_files = data.get_files_dao()
    run = dao_runs.get(run_id)

    if filetype == _FileType.ARTIFACT:
        target_files = run['artifacts']
    elif filetype == _FileType.SOURCE:
        target_files = run['experiment']['sources']
    else:
        raise Exception("Unknown file type: %s" % filetype)

    memory_file = io.BytesIO()
    with zipfile.ZipFile(memory_file, 'w') as zf:
        for f in target_files:
            # source and artifact files use a different data structure
            file_id = f['file_id'] if 'file_id' in f else f[1]
            file, filename, upload_date = dao_files.get(file_id)
            data = zipfile.ZipInfo(filename, date_time=upload_date.timetuple())
            data.compress_type = zipfile.ZIP_DEFLATED
            zf.writestr(data, file.read())
    memory_file.seek(0)

    fn_suffix = _filetype_suffices[filetype]
    return send_file(memory_file, attachment_filename='run{}_{}.zip'.format(run_id, fn_suffix), as_attachment=True)
def get_files_zip ( run_id : int , filetype : _FileType ) : data = current_app . config [ "data" ] dao_runs = data . get_run_dao ( ) dao_files = data . get_files_dao ( ) run = dao_runs . get ( run_id ) if filetype == _FileType . ARTIFACT : target_files = run [ 'artifacts' ] elif filetype == _FileType . SOURCE : target_files = run [ 'experiment' ] [ 'sources' ] else : raise Exception ( "Unknown file type: %s" % filetype ) memory_file = io . BytesIO ( ) with zipfile . ZipFile ( memory_file , 'w' ) as zf : for f in target_files : file_id = f [ 'file_id' ] if 'file_id' in f else f [ 1 ] file , filename , upload_date = dao_files . get ( file_id ) data = zipfile . ZipInfo ( filename , date_time = upload_date . timetuple ( ) ) data . compress_type = zipfile . ZIP_DEFLATED zf . writestr ( data , file . read ( ) ) memory_file . seek ( 0 ) fn_suffix = _filetype_suffices [ filetype ] return send_file ( memory_file , attachment_filename = 'run{}_{}.zip' . format ( run_id , fn_suffix ) , as_attachment = True )
Send all artifacts or sources of a run as ZIP.