Possible API for Sematic Checkpoints

checkpoint_example.py

from sematic import load_checkpoint, store_checkpoint

class MyCheckpoint:
    # ... can be anything you like

@sematic.func
def do_something_really_long(some_val: TheInput) -> TheOutput:
    # Load the checkpoint. It will be None if there is no checkpoint
    # that has been stored. This means that it's impossible to distinguish
    # between a checkpoint whose value is `None`, and an unset checkpoint.
    # That should be fine though, since using `None` as a checkpoint wouldn't
    # really be transferring any useful info anyway.
    checkpoint: Optional[MyCheckpoint] = load_checkpoint()
    # ... perform the work, using the checkpoint if present

    # at any point during the work, if an intermediate result is available,
    # store a checkpoint
    store_checkpoint(checkpoint)
    
    # ... more work
    return the_result

### DIY version, until Sematic implements something like the above:

def store_checkpoint(checkpoint):
    checkpoint_path = f"s3://some-bucket/checkpoints/{sematic.context().run_id}"
    save_to_s3(pickle.dumps(checkpoint), checkpoint_path)

def load_checkpoint():
    checkpoint_path = f"s3://some-bucket/checkpoints/{sematic.context().run_id}"
    if path_exists(checkpoint_path):
        return pickle.loads(read_from_s3(checkpoint_path))
    return None