Proposal for File Transfer

[AlejandroEsquivel]

The initial UX proposal for file transfer is specified below:

# S3 (Not implemented Example)

## pull in file from remote host (source) and place in temp location (destination)
file = ct.files.TransferFrom("s3://file")
@ct.electron(
    files=[file]
)
def task(files):
   # s3://file -> /tmp/{id}
   input_file = files[0] # = /tmp/{id}
   with open(input_file,'r'):
       pass
   pass


## push file from temp location (source) and push file to remote host (destination) 
file = ct.files.TransferTo("s3://file")
@ct.electron(
    files=[file]
)
def task(files):
   output_file = files[0] # = /tmp/{id}
   with open(output_file,'w'):
       pass
   pass
   # /tmp/{id} -> s3://file


# Rsync (Local)

## copy file from (source) and place in new location (destination) [BEFORE - default]
file = ct.files.FileTransfer(from_filepath="/home/shared/file_source", to_filepath="/home/shared/file_dest")
@ct.electron(
    files=[file]
)
def task(files):
   # /home/shared/file_source -> /home/shared/file_dest

   some_file = files[0] # = /home/shared/file_dest
   with open(some_file,'r'):
       pass
   pass


## copy file from (source) and place in new location (destination) [AFTER]
file = ct.files.FileTransfer(from_filepath="/home/shared/file_source", to_filepath="/home/shared/file_dest", order=ct.files.Order.AFTER)
@ct.electron(
    files=[file]
)
def task(files):
   some_file = files[0] # = /home/shared/file_source (or may be none - haven't decided)
   with open(some_file,'r'):
       pass
   pass
# /home/shared/file_source -> /home/shared/file_dest


# Rsync (Remote)

## rsync ssh transfer file from remote (source) and place in temp location in local machine (destination)
file = ct.files.TransferFrom("/Users/user/Documents/source", strategy=Rsync(host='12.12.12.12', user='admin'))
@ct.electron(
    executor='ssh',
    files=[file]
)
def task(files):
   # /Users/user/Documents/source -> /tmp/{id}
   input_file = files[0] # = /tmp/{id}
   with open(input_file,'r'):
       pass
   pass


##  rsync ssh transfer file from local (source) and place in new remote location (destination)
file = ct.files.TransferTo("/Users/user/Documents/dest",strategy=Rsync(host='12.12.12.12', user='admin'))
@ct.electron(
    executor='ssh',
    files=[file]
)
def task(files):
   output_file = files[0] # = /tmp/{id}
   with open(output_file,'w'):
       pass
   pass
# /tmp/{id} -> /Users/user/Documents/dest

[santoshkumarradha]

@AlejandroEsquivel great ! can you also outline the same for "all files" in current directory and other batch process ? like all files in another directory etc.. ?

[santoshkumarradha]

I guess we could do Folder("./") or something ?
The use case is, after a bash like computation, there would be a lot of files (~20-30) that are generated by the electron out of which almost everything would be needed for the next execution. So an easy UX for this is much needed for real world use case.

[AlejandroEsquivel]

@santoshkumarradha I'm still thinking about what a good pattern is to adopt for the folder use case, initially I was thinking we can support globs such as Files('./records/**.csv') to pick out only certain files or use regex to select a subset of files, the simpler case would also just be using File('/records/') where it is implicitly determined the path is referencing a folder.

[santoshkumarradha]

Both works on my end ! 👍

[cjao]

Related question -- are we assuming that the file names are known at workflow construction?

[cjao]

Hi @AlejandroEsquivel, how do these dependencies interact with the electron deps? Which ones are to be handled first?

[santoshkumarradha]

I vote for files to be handled first and then other things as some of the call before can potentially involve changing/modifying files.

[AlejandroEsquivel]

Agreed @santoshkumarradha

[AlejandroEsquivel]

Converted to general discussion around implementation of File Transfer, will post here if there are key decisions around how file transfers will create associations in transport graph

[cjao]

Currently an electron may be invoked several times in a workflow with different arguments. Do we expect the file deps to remain the same for each invocation?

[santoshkumarradha]

beautiful question. There is a future issue to deal with dynamic change in electron dependencies ! For now, we can definitely go with the assumption that electrons are only parameterized by args, kwargs for multiple re-use.

[santoshkumarradha]

One important use case example that adds external implicit dependency while constructing graphs

def create_file():
    with open("my_file",'wb') as b:
        pass


def add(a,i):
    bash(mpirun -np 32 very costly program)


def collect():
    b=0
    for i in range(10):
         with open(f"my_file_{i}",'rb') as b:
            ...
    return b

#workflow 
create_file()
futures=[]
for i in range(10):
    future.append(add(a=i**2,i))
collect()#.dependes(futures) 
add()

[cjao]

Is it unnatural to restructure the code like the following?

def create_file():
    with open("my_file",'wb') as b:
        
    return "my_file"


def add(a,i, uri):
    file_URIs = bash(mpirun -np 32 very costly program)
    return file_URIs


def collect(URIs):
    b=0
    for uri in URIs:
        with open(uri, "rb") as f:
          ...
    return b

#workflow 
uri = create_file()
futures=[]
for i in range(10):
    future.append(add(a=i**2,i, uri))
collect(futures)

[santoshkumarradha]

Updoot for the new Ux !