magic dir

magicdir.py

from metaflow import FlowSpec, step
from functools import wraps

from functools import wraps

dir = 'mydir'

def magicdir(f):
    artifact = 'magicdir'
    @wraps(f)
    def func(self):
        from io import BytesIO
        from tarfile import TarFile
        existing = getattr(self, artifact, None)
        if existing:
            buf = BytesIO(existing)
            with TarFile(mode='r', fileobj=buf) as tar:
                tar.extractall()
        f(self)
        buf = BytesIO()
        with TarFile(mode='w', fileobj=buf) as tar:
            tar.add(dir)
        setattr(self, artifact, buf.getvalue())
    return func

class MagicDirFlow(FlowSpec):

    @magicdir
    @step
    def start(self):
        with open('mydir/output1', 'w') as f:
            f.write('hello world')
        with open('mydir/output2', 'w') as f:
            f.write('hello world again')
        self.next(self.end)

    @magicdir
    @step
    def end(self):
        print('first', open('mydir/output1').read())
        print('second', open('mydir/output1').read())

if __name__ == "__main__":
    MagicDirFlow()