local

State client for the local filesystem.

LocalStateClient

Bases: StateClient

Read and write state as JSON files on the local filesystem.

The path argument is interpreted as a filesystem path (absolute or relative).

Source code in src/cloe_delta_table_manager/state/clients/local.py

class LocalStateClient(StateClient):
    """Read and write state as JSON files on the local filesystem.

    The ``path`` argument is interpreted as a filesystem path (absolute or relative).
    """

    def read_state(self, path: str, state_type: type[StateT]) -> StateT:
        """Read state from a local JSON file.

        Args:
            path: Filesystem path to the JSON state file.
            state_type: The concrete State subclass to deserialize into.

        Returns:
            A State instance loaded from the file.

        Raises:
            FileNotFoundError: If the file does not exist.

        """
        file_path = Path(path)
        if not file_path.exists():
            raise FileNotFoundError(f"State file not found: {path}")
        return state_type.model_validate_json(file_path.read_text())

    def write_state(self, path: str, state: State) -> None:
        """Write state to a local JSON file.

        Parent directories are created automatically if they do not exist.
        The version in the state metadata is incremented before writing.

        Args:
            path: Filesystem path for the JSON state file.
            state: The State instance to persist.

        """
        file_path = Path(path)
        state.metadata.version += 1
        state.metadata.last_updated = datetime.now()
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(state.model_dump_json(indent=2))

    def state_exists(self, path: str) -> bool:
        """Check whether a state file exists on the local filesystem.

        Args:
            path: Filesystem path to check.

        Returns:
            True if the file exists, False otherwise.

        """
        return Path(path).exists()

read_state(path, state_type)

Read state from a local JSON file.

Parameters:

Name	Type	Description	Default
path	str	Filesystem path to the JSON state file.	required
state_type	type[StateT]	The concrete State subclass to deserialize into.	required

Returns:

Type	Description
StateT	A State instance loaded from the file.

Raises:

Type	Description
FileNotFoundError	If the file does not exist.

Source code in src/cloe_delta_table_manager/state/clients/local.py

def read_state(self, path: str, state_type: type[StateT]) -> StateT:
    """Read state from a local JSON file.

    Args:
        path: Filesystem path to the JSON state file.
        state_type: The concrete State subclass to deserialize into.

    Returns:
        A State instance loaded from the file.

    Raises:
        FileNotFoundError: If the file does not exist.

    """
    file_path = Path(path)
    if not file_path.exists():
        raise FileNotFoundError(f"State file not found: {path}")
    return state_type.model_validate_json(file_path.read_text())

state_exists(path)

Check whether a state file exists on the local filesystem.

Parameters:

Name	Type	Description	Default
path	str	Filesystem path to check.	required

Returns:

Type	Description
bool	True if the file exists, False otherwise.

Source code in src/cloe_delta_table_manager/state/clients/local.py

def state_exists(self, path: str) -> bool:
    """Check whether a state file exists on the local filesystem.

    Args:
        path: Filesystem path to check.

    Returns:
        True if the file exists, False otherwise.

    """
    return Path(path).exists()

write_state(path, state)

Write state to a local JSON file.

Parent directories are created automatically if they do not exist. The version in the state metadata is incremented before writing.

Parameters:

Name	Type	Description	Default
path	str	Filesystem path for the JSON state file.	required
state	State	The State instance to persist.	required

Source code in src/cloe_delta_table_manager/state/clients/local.py

def write_state(self, path: str, state: State) -> None:
    """Write state to a local JSON file.

    Parent directories are created automatically if they do not exist.
    The version in the state metadata is incremented before writing.

    Args:
        path: Filesystem path for the JSON state file.
        state: The State instance to persist.

    """
    file_path = Path(path)
    state.metadata.version += 1
    state.metadata.last_updated = datetime.now()
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(state.model_dump_json(indent=2))