Use case save/restore with Pythonflow for long running data flow operations.

test_pythonflow.py

"""
Use case save/restore with Pythonflow for long running data flow operations.
Pythonflow is a very lighter alternative to tensorflow's graphs.

Prerequisite:
$ pip install pythonflow

In this example, we model round(a+b) with two save points.

Our file-database will have the structure:
📂 db
   - processing.txt
   - post_processing.txt

"""

from pathlib import Path
import pythonflow as pf


db: Path = Path(__file__).parent.parent / 'db'


@pf.opmethod(length=1)
def math_round(x):
    return round(x)


@pf.opmethod()
def save_point(x, db_name: str):
    print(f'Saving to {db_name}: {x}')
    (db / f'{db_name}.txt').write_text(str(x))
    return x


def restore(db_name: str) -> float:
    print(f'Restore from {db_name}')
    return float((db / f'{db_name}.txt').read_text())
"""
Define graph as
 a
  ⇘
   processing ⇒ saved_processing ⇒ saved_post_processing
  ⇗
 b
"""
with pf.Graph() as graph:
    a = pf.constant(name='a')
    b = pf.constant(name='b')
    processing = a + b  # Super expensive computation
    saved_processing = save_point(processing, 'processing')
    post_processing = math_round(saved_processing)
    saved_post_processing = save_point(post_processing, 'post_processing')

# Execute processing step, no save points.
result = graph(processing, {'a': 8, 'b': 1.7})
print(f'Result is {result}')

# Execute whole graph.
result = graph(saved_post_processing, {'a': 8, 'b': 1.7})
print(f'Result is {result}')

# Only execute post processing step by restoring processing step.
result = graph(post_processing, {saved_processing: restore('processing')})
print(f'Result is {result}')

# Result is 9.7
# Saving to processing: 9.7
# Saving to post_processing: 10
# Result is 10
# Restore from processing
# Result is 10