proxystore.connectors.file¶

proxystore/connectors/file.py

File system connector implementation.

FileKey ¶

Bases: NamedTuple

Key to objects in a file system directory.

Attributes:

filename (str) –

Unique object filename.

FileConnector ¶

FileConnector(
    store_dir: str | PathLike[str],
    *,
    clear: bool = True,
    buffering: int = -1,
    dir_mode: int = 448,
    file_mode: int = 384,
)

Connector to shared file system.

This connector writes objects to unique files within store_dir. Each object is written to a hidden temporary file in store_dir which is then renamed, so an object's file only exists once it is completely written.

Parameters:

store_dir (str | PathLike[str]) –

Path to directory to store data in. If the connector creates this directory, it is removed when the connector is closed with clear=True (the default), such as by the owner store. A directory which already existed is never removed.
clear (bool, default: True ) –

Clear all objects on close() by removing store_dir if the connector created it.
buffering (int, default: -1 ) –

Buffering policy used with open().
dir_mode (int, default: 448 ) –

Permissions of store_dir if the connector creates it. The default only gives the current user access. An existing store_dir keeps its permissions, and parent directories are created with the default permissions.
file_mode (int, default: 384 ) –

Permissions of the object files. The default only gives the current user access.

Note

dir_mode and file_mode are applied exactly and are not limited by the umask, so, e.g., dir_mode=0o770, file_mode=0o660 shares the objects with the group of store_dir.

Source code in proxystore/connectors/file.py

def __init__(
    self,
    store_dir: str | os.PathLike[str],
    *,
    clear: bool = True,
    buffering: int = -1,
    dir_mode: int = 0o700,
    file_mode: int = 0o600,
) -> None:
    self.store_dir = os.path.abspath(store_dir)
    self.clear = clear
    self.buffering = buffering
    self.dir_mode = dir_mode
    self.file_mode = file_mode
    # Only a directory this connector created is removed when cleared,
    # so the files of a directory which already existed are never lost.
    self._created_dir = False

    if not os.path.exists(self.store_dir):
        os.makedirs(os.path.dirname(self.store_dir), exist_ok=True)
        try:
            os.mkdir(self.store_dir, mode=self.dir_mode)
        except FileExistsError:
            # Another process created the directory first.
            pass
        else:
            os.chmod(self.store_dir, self.dir_mode)
            self._created_dir = True

close ¶

close(*, clear: bool | None = None) -> bool

Close the connector and clean up.

Warning

This will delete the store_dir directory by default if the connector created it. A store_dir which already existed is left as is, including its objects, and a warning is logged.

Note

A Store only clears its connector when closed if the store is the owner. Stores created implicitly, such as when resolving a proxy, never clear their connector.

Warning

This method should only be called at the end of the program when the connector will no longer be used, for example once all proxies have been resolved.

Parameters:

clear (bool | None, default: None ) –

Remove the store directory if the connector created it. Overrides the default value of clear provided when the FileConnector was instantiated.

Returns:

bool –

If the store directory was removed.

Source code in proxystore/connectors/file.py

def close(self, *, clear: bool | None = None) -> bool:
    """Close the connector and clean up.

    Warning:
        This will delete the `store_dir` directory by default if the
        connector created it. A `store_dir` which already existed is
        left as is, including its objects, and a warning is logged.

    Note:
        A [`Store`][proxystore.store.base.Store] only clears its
        connector when closed if the store is the
        [`owner`][proxystore.store.base.Store.owner]. Stores created
        implicitly, such as when resolving a proxy, never clear their
        connector.

    Warning:
        This method should only be called at the end of the program
        when the connector will no longer be used, for example once all
        proxies have been resolved.

    Args:
        clear: Remove the store directory if the connector created
            it. Overrides the default
            value of `clear` provided when the
            [`FileConnector`][proxystore.connectors.file.FileConnector]
            was instantiated.

    Returns:
        If the store directory was removed.
    """
    clear = self.clear if clear is None else clear
    if not clear:
        return False
    if not self._created_dir:
        if os.path.isdir(self.store_dir):
            logger.warning(
                'Not removing %s because it existed before the '
                'FileConnector was created. Use a new directory or '
                'close with clear=False.',
                self.store_dir,
            )
        return False
    shutil.rmtree(self.store_dir, ignore_errors=True)
    return True

config ¶

config() -> dict[str, Any]

Get the connector configuration.

The configuration contains all the information needed to reconstruct the connector object.

Source code in proxystore/connectors/file.py

def config(self) -> dict[str, Any]:
    """Get the connector configuration.

    The configuration contains all the information needed to reconstruct
    the connector object.
    """
    return {
        'store_dir': self.store_dir,
        'clear': self.clear,
        'buffering': self.buffering,
        'dir_mode': self.dir_mode,
        'file_mode': self.file_mode,
    }

from_config `classmethod` ¶

from_config(config: dict[str, Any]) -> FileConnector

Create a new connector instance from a configuration.

Parameters:

config (dict[str, Any]) –

Configuration returned by .config().

Source code in proxystore/connectors/file.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> FileConnector:
    """Create a new connector instance from a configuration.

    Args:
        config: Configuration returned by `#!python .config()`.
    """
    return cls(**init_kwargs(cls, config))

evict ¶

evict(key: FileKey) -> None

Evict the object associated with the key.

Parameters:

key (FileKey) –

Key associated with object to evict.

Source code in proxystore/connectors/file.py

def evict(self, key: FileKey) -> None:
    """Evict the object associated with the key.

    Args:
        key: Key associated with object to evict.
    """
    # The file may already have been removed by a concurrent evict().
    with contextlib.suppress(FileNotFoundError):
        os.remove(self._path(key))

exists ¶

exists(key: FileKey) -> bool

Check if an object associated with the key exists.

Parameters:

key (FileKey) –

Key potentially associated with stored object.

Returns:

bool –

If an object associated with the key exists.

Source code in proxystore/connectors/file.py

def exists(self, key: FileKey) -> bool:
    """Check if an object associated with the key exists.

    Args:
        key: Key potentially associated with stored object.

    Returns:
        If an object associated with the key exists.
    """
    return os.path.exists(self._path(key))

get ¶

get(key: FileKey) -> BytesLike | None

Get the serialized object associated with the key.

Parameters:

key (FileKey) –

Key associated with the object to retrieve.

Returns:

BytesLike | None –

Serialized object or None if the object does not exist.

Source code in proxystore/connectors/file.py

def get(self, key: FileKey) -> BytesLike | None:
    """Get the serialized object associated with the key.

    Args:
        key: Key associated with the object to retrieve.

    Returns:
        Serialized object or `None` if the object does not exist.
    """
    try:
        with open(self._path(key), 'rb', buffering=self.buffering) as f:
            return f.read()
    except FileNotFoundError:
        return None

get_batch ¶

get_batch(
    keys: Sequence[FileKey],
) -> list[BytesLike | None]

Get a batch of serialized objects associated with the keys.

Parameters:

keys (Sequence[FileKey]) –

Sequence of keys associated with objects to retrieve.

Returns:

list[BytesLike | None] –

List with same order as keys with the serialized objects or
list[BytesLike | None] –

None if the corresponding key does not have an associated object.

Source code in proxystore/connectors/file.py

def get_batch(self, keys: Sequence[FileKey]) -> list[BytesLike | None]:
    """Get a batch of serialized objects associated with the keys.

    Args:
        keys: Sequence of keys associated with objects to retrieve.

    Returns:
        List with same order as `keys` with the serialized objects or
        `None` if the corresponding key does not have an associated object.
    """
    return [self.get(key) for key in keys]

new_key ¶

new_key(obj: BytesLike | None = None) -> FileKey

Create a new key.

Parameters:

obj (BytesLike | None, default: None ) –

Optional object which the key will be associated with. Ignored by this implementation.

Returns:

FileKey –

Key which can be used to retrieve an object once set() has been called on the key.

Source code in proxystore/connectors/file.py

def new_key(self, obj: BytesLike | None = None) -> FileKey:
    """Create a new key.

    Args:
        obj: Optional object which the key will be associated with.
            Ignored by this implementation.

    Returns:
        Key which can be used to retrieve an object once \
        [`set()`][proxystore.connectors.file.FileConnector.set] \
        has been called on the key.
    """
    return FileKey(filename=str(uuid.uuid4()))

put ¶

put(obj: BytesLike) -> FileKey

Put a serialized object in the store.

Parameters:

obj (BytesLike) –

Serialized object to put in the store.

Returns:

FileKey –

Key which can be used to retrieve the object.

Source code in proxystore/connectors/file.py

def put(self, obj: BytesLike) -> FileKey:
    """Put a serialized object in the store.

    Args:
        obj: Serialized object to put in the store.

    Returns:
        Key which can be used to retrieve the object.
    """
    key = FileKey(filename=str(uuid.uuid4()))
    self.set(key, obj)
    return key

put_batch ¶

put_batch(objs: Sequence[BytesLike]) -> list[FileKey]

Put a batch of serialized objects in the store.

Parameters:

objs (Sequence[BytesLike]) –

Sequence of serialized objects to put in the store.

Returns:

list[FileKey] –

List of keys with the same order as objs which can be used to
list[FileKey] –

retrieve the objects.

Source code in proxystore/connectors/file.py

def put_batch(self, objs: Sequence[BytesLike]) -> list[FileKey]:
    """Put a batch of serialized objects in the store.

    Args:
        objs: Sequence of serialized objects to put in the store.

    Returns:
        List of keys with the same order as `objs` which can be used to
        retrieve the objects.
    """
    return [self.put(obj) for obj in objs]

set ¶

set(key: FileKey, obj: BytesLike) -> None

Set the object associated with a key.

Note

The Connector provides write-once, read-many semantics. Thus, set() should only be called once per key, otherwise unexpected behavior can occur.

Parameters:

key (FileKey) –

Key that the object will be associated with.
obj (BytesLike) –

Object to associate with the key.

Source code in proxystore/connectors/file.py

def set(self, key: FileKey, obj: BytesLike) -> None:
    """Set the object associated with a key.

    Note:
        The [`Connector`][proxystore.connectors.protocols.Connector]
        provides write-once, read-many semantics. Thus,
        [`set()`][proxystore.connectors.file.FileConnector.set]
        should only be called once per key, otherwise unexpected behavior
        can occur.

    Args:
        key: Key that the object will be associated with.
        obj: Object to associate with the key.
    """
    # Write to a unique temporary file and rename it so readers never
    # see a partially written file, and concurrent writers of the same
    # key do not write to the same file.
    tmp = os.path.join(
        self.store_dir,
        f'.{key.filename}.{uuid.uuid4().hex}.tmp',
    )
    try:
        with open(
            tmp,
            'xb',
            buffering=self.buffering,
            opener=self._opener,
        ) as f:
            os.fchmod(f.fileno(), self.file_mode)
            f.write(obj)
        os.replace(tmp, self._path(key))
    except BaseException:
        with contextlib.suppress(FileNotFoundError):
            os.remove(tmp)
        raise

proxystore.connectors.file¶

FileKey ¶

FileConnector ¶

close ¶

config ¶

from_config classmethod ¶

evict ¶

exists ¶

get ¶

get_batch ¶

new_key ¶

put ¶

put_batch ¶

set ¶

from_config `classmethod` ¶