BaseCheckpoint
Base type for task checkpoint helpers. Subclasses load prior checkpoint data from prev_checkpoint into a local workspace and upload new state to checkpoint_path.
Attributes
| Attribute | Type | Description |
|---|---|---|
| path | pathlib.Path | Local directory for reading and writing checkpoint files (your format). |
Methods
path()
@classmethod
def path() - > pathlib.Path
Local directory for reading and writing checkpoint files (your format).
Returns
| Type | Description |
|---|---|
pathlib.Path | The local filesystem path where checkpoint files are stored and managed |
prev_exists()
@classmethod
def prev_exists() - > bool
Whether the runtime provided a previous-checkpoint prefix (retry / resume).
Returns
| Type | Description |
|---|---|
bool | True if a previous checkpoint exists and is available for restoration, False otherwise |
load()
@classmethod
def load() - > Optional[pathlib.Path]
Load checkpoint data from the remote source (async).
Returns
| Type | Description |
|---|---|
Optional[pathlib.Path] | The path to the loaded checkpoint directory, or None if no checkpoint was found |
save()
@classmethod
def save(
data: pathlib.Path | str | bytes
) - > None
Save checkpoint data to the remote destination (async).
Parameters
| Name | Type | Description |
|---|---|---|
| data | `pathlib.Path | str |
Returns
| Type | Description |
|---|---|
None | Nothing is returned |
load_sync()
@classmethod
def load_sync() - > Optional[pathlib.Path]
Load checkpoint data from the remote source (sync task code).
Returns
| Type | Description |
|---|---|
Optional[pathlib.Path] | The path to the loaded checkpoint directory, or None if no checkpoint was found |
save_sync()
@classmethod
def save_sync(
data: pathlib.Path | str | bytes
) - > None
Save checkpoint data to the remote destination (sync task code).
Parameters
| Name | Type | Description |
|---|---|---|
| data | `pathlib.Path | str |
Returns
| Type | Description |
|---|---|
None | Nothing is returned |