proxystore.store¶

proxystore/store/__init__.py

The ProxyStore Store interface.

Every Store is registered in a registry that is global to the Python process and keyed by the unique id of the store. A store is registered when it is created and unregistered when it is closed. When a proxy is resolved, the store which created the proxy is looked up in the registry, or initialized and registered from the configuration contained in the proxy if the store does not exist in the process.

Store ¶

Store(
    connector: ConnectorT,
    *,
    name: str | None = None,
    serializer: SerializerT | None = None,
    deserializer: DeserializerT | None = None,
    cache_size_bytes: int = DEFAULT_CACHE_SIZE_BYTES,
    cache_mode: CacheModeT = 'objects',
    metrics: bool | StoreMetrics = False,
    populate_target: bool = True,
    owner: bool = True,
    _id: str | None = None,
)

Bases: Generic[ConnectorT]

Key-value store interface for proxies.

Tip

A Store instance can be used as a context manager which will automatically call close() on exit.

with Store(...) as store:
    key = store.put('value')
    store.get(key)

Warning

The default value of populate_target=True can cause unexpected behavior when the deserializer of the store does not undo its serializer because neither is applied to the target object cached in the resulting Proxy.

import pickle
from proxystore.store import Store
from proxystore.connectors.local import LocalConnector

with Store(
    LocalConnector(),
    serializer=lambda s: s,
    deserializer=pickle.loads,
) as store:
    data = [1, 2, 3]
    data_bytes = pickle.dumps(data)

    data_proxy = store.proxy(data_bytes, populate_target=True)

    print(data_proxy)
    # b'\x80\x04\x95\x0b\x00\x00\x00\x00\x00\x00\x00]\x94(K\x01K\x02K\x03e.'

In this example, the serialized data_bytes was populated as the target object in the resulting proxy so the proxy looks like a proxy of bytes rather than the intended list of integers. To fix this, set populate_target=False so the deserializer is correctly applied to data_bytes when the proxy is resolved.

Note

This class is thread-safe. Different keys are fetched from the connector at the same time, and a key that is already being fetched by another thread is not fetched again. The thread waits for that fetch instead. The connector must be thread-safe (see Connector).

Note

Each store has a unique id which is used to register the store in a registry that is global to the Python process. The store is registered when initialized and unregistered when close() is called. Proxies created by a store contain the configuration of the store, including the ID, so the store can be found in the registry, or initialized if needed, when a proxy is resolved.

Warning

This class cannot be pickled. If you need to recreate a Store within another process, share a StoreConfig, a serializable and pickle-compatible type, that can be created using Store.config().

To reconstruct the instance from the config, use get_or_create_store(), which reuses the store if it already exists in the process, or Store.from_config(), which always creates a new store instance.

Parameters:

connector (ConnectorT) –

Connector instance to use for object storage. Keys taken and returned by the store's methods (KeyT) have the key type of the connector (e.g., a FileKey for a FileConnector).
name (str | None, default: None ) –

Optional name of the store used in logs and error messages. The name does not need to be unique.
serializer (SerializerT | None, default: None ) –

Optional callable which serializes every object put in the store. If None, the default serializer (serialize()) will be used.
deserializer (DeserializerT | None, default: None ) –

Optional callable which deserializes every object gotten from the store, including when proxies are resolved. If None, the default deserializer (deserialize()) will be used. The serializer and deserializer are part of the config() of the store, so they must be importable in other processes.
cache_size_bytes (int, default: DEFAULT_CACHE_SIZE_BYTES ) –

Maximum size of the LRU cache in bytes (256 MiB by default). Each entry counts as the size of the object's serialized data, so with the 'objects' cache mode this is an estimate of the memory used. Objects larger than this are not cached. If 0, the cache is disabled. The cache is local to the Python process.
cache_mode (CacheModeT, default: 'objects' ) –

What the cache holds. With 'objects', the cache holds deserialized objects, so a cache hit is free but every get() and proxy of a key in this process shares the same object, and changing it changes all of them. With 'bytes', the cache holds the serialized data and deserializes it on every hit, so each caller gets its own object at the cost of deserializing each time.
metrics (bool | StoreMetrics, default: False ) –

Enable recording operation metrics. Pass a StoreMetrics instance to configure it (e.g., max_keys). The configuration of StoreMetrics is not part of the store's config(), so stores created from the config in other processes use the defaults.
populate_target (bool, default: True ) –

Set the default value of populate_target for proxy methods of the store.
owner (bool, default: True ) –

This store owns the objects stored by the connector, so the connector is cleared (e.g., the directory of a FileConnector is deleted), according to the connector's clear setting, when this store is closed. Stores created implicitly, such as when a proxy is resolved in another process, are never owners.

Raises:

ValueError –

If cache_size_bytes is less than zero or cache_mode is not 'objects' or 'bytes'.

Source code in proxystore/store/base.py

def __init__(
    self,
    connector: ConnectorT,
    *,
    name: str | None = None,
    serializer: SerializerT | None = None,
    deserializer: DeserializerT | None = None,
    cache_size_bytes: int = DEFAULT_CACHE_SIZE_BYTES,
    cache_mode: CacheModeT = 'objects',
    metrics: bool | StoreMetrics = False,
    populate_target: bool = True,
    owner: bool = True,
    _id: str | None = None,
) -> None:
    # _id is private and only used by from_config() to recreate a store
    # with the same ID. Users should not set it because stores with the
    # same ID are considered the same store.
    if cache_size_bytes < 0:
        raise ValueError(
            f'Cache size cannot be negative. Got {cache_size_bytes}.',
        )
    if cache_mode not in ('objects', 'bytes'):
        raise ValueError(
            "Cache mode must be 'objects' or 'bytes'. "
            f'Got {cache_mode!r}.',
        )

    self._connector = connector
    self._cache: LRUCache[ConnectorKeyT, Any] = LRUCache(
        cache_size_bytes,
    )
    self._id = _id if _id is not None else uuid.uuid4().hex
    self._name = name
    self._metrics: StoreMetrics | None
    if isinstance(metrics, StoreMetrics):
        self._metrics = metrics
    else:
        self._metrics = StoreMetrics() if metrics else None
    self._cache_size_bytes = cache_size_bytes
    self._cache_mode = cache_mode
    self._serializer = serializer
    self._deserializer = deserializer
    self._populate_target = populate_target
    self._owner = owner
    # A proxy made by this store holds its configuration, so the
    # configuration is made once rather than for every proxy.
    self._config = self._make_config()
    _stores.add(self)
    registry.register(self)

    logger.info('Initialized %s', self)

connector `property` ¶

connector: ConnectorT

Connector used by this store for object storage.

cache `property` ¶

cache: LRUCache[ConnectorKeyT, Any]

Cache of objects (or serialized data) in this process.

id `property` ¶

id: str

Unique ID of this Store.

The ID is shared by stores initialized from the config() of this store (e.g., when a proxy created by this store is resolved in another process).

owner `property` ¶

owner: bool

This store owns the objects stored by its connector.

Only the owner clears the connector when closed. See close().

name `property` ¶

name: str | None

Optional name of this Store.

metrics `property` ¶

metrics: StoreMetrics | None

Optional metrics for this instance.

serializer `property` ¶

serializer: SerializerT

Serializer for this instance.

deserializer `property` ¶

deserializer: DeserializerT

Deserializer for this instance.

close ¶

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

Close the connector associated with the store.

This will (1) close the connector and (2) unregister the store. If this store is the owner and the connector removed the stored objects, the ID of the store stays closed for the rest of the process, so resolving proxies of the store in this process raises a StoreClosedError instead of making a new store.

Warning

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

Parameters:

clear (bool | None, default: None ) –

Clear the objects stored by the connector (see Connector.close()). If None, the connector's default is used if this store is the owner and the connector is not cleared otherwise.

Source code in proxystore/store/base.py

def close(self, *, clear: bool | None = None) -> None:
    """Close the connector associated with the store.

    This will (1) close the connector and (2) unregister the store.
    If this store is the [`owner`][proxystore.store.base.Store.owner]
    and the connector removed the stored objects, the ID of the store
    stays closed for the rest of the process, so resolving proxies of
    the store in this process raises a
    [`StoreClosedError`][proxystore.store.exceptions.StoreClosedError]
    instead of making a new store.

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

    Args:
        clear: Clear the objects stored by the connector (see
            [`Connector.close()`][proxystore.connectors.protocols.Connector.close]).
            If `None`, the connector's default is used if this store is
            the [`owner`][proxystore.store.base.Store.owner] and the
            connector is not cleared otherwise.
    """
    # Deleted owned proxies of this store may still be waiting to evict
    # their objects, which needs this store to be registered.
    evictions.wait()

    if clear is None and not self.owner:
        clear = False

    cleared: bool | None = None
    try:
        if clear is None:
            cleared = self.connector.close()
        else:
            cleared = self.connector.close(clear=clear)
    finally:
        registry.unregister(self, closed=self.owner and cleared is True)

config ¶

config() -> StoreConfig

Get the store configuration.

Example

>>> store = Store(...)
>>> config = store.config()
>>> store = Store.from_config(config)

Note

The configuration is made when the store is created, and every call returns the same (immutable) object. Changes made to the connector after the store is created are not included.

Returns:

StoreConfig –

Store configuration.

Source code in proxystore/store/base.py

def config(self) -> StoreConfig:
    """Get the store configuration.

    Example:
        ```python
        >>> store = Store(...)
        >>> config = store.config()
        >>> store = Store.from_config(config)
        ```

    Note:
        The configuration is made when the store is created, and every
        call returns the same (immutable) object. Changes made to the
        connector after the store is created are not included.

    Returns:
        Store configuration.
    """
    return self._config

from_config `classmethod` ¶

from_config(
    config: StoreConfig, *, owner: bool | None = None
) -> Store[Any]

Create a new store instance from a configuration.

The new store has the same id as the store which produced the configuration. If config does not contain an ID (e.g., a configuration loaded from a TOML file), a new ID is generated.

By default, the new store is the owner only if config does not contain an ID. A config with an ID came from an existing store, and that store is the owner. A config without an ID describes a new store, so the new store is the owner.

Tip

Use get_or_create_store() to reuse a store with the same ID that already exists in this process.

Parameters:

config (StoreConfig) –

Configuration returned by .config().
owner (bool | None, default: None ) –

The new store owns the objects stored by the connector. See the owner parameter of Store. If None, the new store is the owner only if config does not contain an ID.

Returns:

Store[Any] –

Store instance.

Source code in proxystore/store/base.py

@classmethod
def from_config(
    cls,
    config: StoreConfig,
    *,
    owner: bool | None = None,
) -> Store[Any]:
    """Create a new store instance from a configuration.

    The new store has the same [`id`][proxystore.store.base.Store.id]
    as the store which produced the configuration. If `config` does not
    contain an ID (e.g., a configuration loaded from a TOML file),
    a new ID is generated.

    By default, the new store is the
    [`owner`][proxystore.store.base.Store.owner] only if `config` does
    not contain an ID. A config with an ID came from an existing store,
    and that store is the owner. A config without an ID describes a new
    store, so the new store is the owner.

    Tip:
        Use [`get_or_create_store()`][proxystore.store.get_or_create_store]
        to reuse a store with the same ID that already exists in this
        process.

    Args:
        config: Configuration returned by `#!python .config()`.
        owner: The new store owns the objects stored by the connector.
            See the `owner` parameter of
            [`Store`][proxystore.store.base.Store]. If `None`, the new
            store is the owner only if `config` does not contain an ID.

    Returns:
        Store instance.
    """
    if owner is None:
        owner = config.id is None
    connector = cast(ConnectorT, config.connector.get_connector())
    return cls(
        connector,
        name=config.name,
        serializer=config.serializer,
        deserializer=config.deserializer,
        cache_size_bytes=config.cache_size_bytes,
        cache_mode=config.cache_mode,
        metrics=config.metrics,
        populate_target=config.populate_target,
        owner=owner,
        _id=config.id,
    )

future ¶

future(
    *,
    evict: bool = False,
    polling: PollingPolicy | None = None,
) -> ProxyFuture[T]

Create a future to an object.

Example

from proxystore.connectors.file import FileConnector
from proxystore.store import Store
from proxystore.store.future import ProxyFuture

def remote_foo(future: ProxyFuture) -> None:
    # Computation that generates a result value needed by
    # the remote_bar function.
    future.set_result(...)

def remote_bar(data: Any) -> None:
    # Function uses data, which is a proxy, as normal, blocking
    # until the remote_foo function has called set_result.
    ...

with Store(FileConnector(...)) as store:
    future = store.future()

    # The invoke_remote function invokes a provided function
    # on a remote process. For example, this could be a serverless
    # function execution.
    foo_result_future = invoke_remote(remote_foo, future)
    bar_result_future = invoke_remote(remote_bar, future.proxy())

    foo_result_future.result()
    bar_result_future.result()

Warning

This method only works if the connector is of type DeferrableConnector.

Parameters:

evict (bool, default: False ) –

If a proxy returned by ProxyFuture.proxy() should evict the object once resolved.
polling (PollingPolicy | None, default: None ) –

Policy for polling the store for the result of the future. If None, the default PollingPolicy is used.

Returns:

ProxyFuture[T] –

Future which can be used to get the result object at a later time or create a proxy which will resolve to the result of the future.

Raises:

NotImplementedError –

If the connector is not of type DeferrableConnector.

Source code in proxystore/store/base.py

def future(
    self,
    *,
    evict: bool = False,
    polling: PollingPolicy | None = None,
) -> ProxyFuture[T]:
    """Create a future to an object.

    Example:
        ```python
        from proxystore.connectors.file import FileConnector
        from proxystore.store import Store
        from proxystore.store.future import ProxyFuture

        def remote_foo(future: ProxyFuture) -> None:
            # Computation that generates a result value needed by
            # the remote_bar function.
            future.set_result(...)

        def remote_bar(data: Any) -> None:
            # Function uses data, which is a proxy, as normal, blocking
            # until the remote_foo function has called set_result.
            ...

        with Store(FileConnector(...)) as store:
            future = store.future()

            # The invoke_remote function invokes a provided function
            # on a remote process. For example, this could be a serverless
            # function execution.
            foo_result_future = invoke_remote(remote_foo, future)
            bar_result_future = invoke_remote(remote_bar, future.proxy())

            foo_result_future.result()
            bar_result_future.result()
        ```

    Warning:
        This method only works if the `connector` is of type
        [`DeferrableConnector`][proxystore.connectors.protocols.DeferrableConnector].

    Args:
        evict: If a proxy returned by
            [`ProxyFuture.proxy()`][proxystore.store.future.ProxyFuture.proxy]
            should evict the object once resolved.
        polling: Policy for polling the store for the result of the
            future. If `None`, the default
            [`PollingPolicy`][proxystore.store.future.PollingPolicy] is
            used.

    Returns:
        Future which can be used to get the result object at a later time \
        or create a proxy which will resolve to the result of the future.

    Raises:
        NotImplementedError: If the `connector` is not of type
            [`DeferrableConnector`][proxystore.connectors.protocols.DeferrableConnector].
    """
    timer = Timer().start()

    if not isinstance(self.connector, DeferrableConnector):
        raise NotImplementedError(
            'The provided connector is type '
            f'{type(self.connector).__name__} which does not implement '
            f'the {DeferrableConnector.__name__} necessary to use the '
            f'{ProxyFuture.__name__} interface.',
        )

    with Timer() as connector_timer:
        key = self.connector.new_key()

    if self.metrics is not None:
        ctime = connector_timer.elapsed_ms
        self.metrics.add_time('store.future.connector', key, ctime)

    factory: PollingStoreFactory[ConnectorT, T] = PollingStoreFactory(
        key,
        store_config=self.config(),
        evict=evict,
        polling=polling,
    )
    future = ProxyFuture(factory)

    timer.stop()
    if self.metrics is not None:
        self.metrics.add_time('store.future', key, timer.elapsed_ms)

    logger.debug(
        'Store(%s): FUTURE %s in %.3f ms',
        self._label,
        key,
        timer.elapsed_ms,
    )
    return future

evict ¶

evict(key: KeyT) -> None

Evict the object associated with the key.

Parameters:

key (KeyT) –

Key associated with object to evict.

Source code in proxystore/store/base.py

def evict(self: Store[Connector[KeyT]], key: KeyT) -> None:
    """Evict the object associated with the key.

    Args:
        key: Key associated with object to evict.
    """
    timer = Timer().start()

    with Timer() as connector_timer:
        self.connector.evict(key)

    if self.metrics is not None:
        ctime = connector_timer.elapsed_ms
        self.metrics.add_time('store.evict.connector', key, ctime)

    self.cache.evict(key)

    timer.stop()
    if self.metrics is not None:
        self.metrics.add_time('store.evict', key, timer.elapsed_ms)

    logger.debug(
        'Store(%s): EVICT %s in %.3f ms',
        self._label,
        key,
        timer.elapsed_ms,
    )

exists ¶

exists(key: KeyT) -> bool

Check if an object associated with the key exists.

Parameters:

key (KeyT) –

Key potentially associated with stored object.

Returns:

bool –

If an object associated with the key exists.

Source code in proxystore/store/base.py

def exists(self: Store[Connector[KeyT]], key: KeyT) -> 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.
    """
    timer = Timer().start()

    res = self.cache.exists(key)
    if not res:
        with Timer() as connector_timer:
            res = self.connector.exists(key)

        if self.metrics is not None:
            ctime = connector_timer.elapsed_ms
            self.metrics.add_time('store.exists.connector', key, ctime)

    timer.stop()
    if self.metrics is not None:
        self.metrics.add_time('store.exists', key, timer.elapsed_ms)

    logger.debug(
        'Store(%s): EXISTS %s in %.3f ms',
        self._label,
        key,
        timer.elapsed_ms,
    )
    return res

get ¶

get(
    key: KeyT, *, default: object | None = None
) -> Any | None

Get the object associated with the key.

Tip

Like dict.get(), None is returned if the object does not exist, so a missing object cannot be distinguished from an object which is None. Use a sentinel default or exists() if you need to distinguish the two.

missing = object()
obj = store.get(key, default=missing)
if obj is missing:
    ...

Parameters:

key (KeyT) –

Key associated with the object to retrieve.
default (object | None, default: None ) –

An optional value to be returned if an object associated with the key does not exist.

Returns:

Any | None –

Object or default if the object does not exist.

Raises:

SerializationError –

If an exception is caught when deserializing the object associated with the key.

Source code in proxystore/store/base.py

def get(
    self: Store[Connector[KeyT]],
    key: KeyT,
    *,
    default: object | None = None,
) -> Any | None:
    """Get the object associated with the key.

    Tip:
        Like [`dict.get()`][dict.get], `None` is returned if the object
        does not exist, so a missing object cannot be distinguished from
        an object which is `None`. Use a sentinel `default` or
        [`exists()`][proxystore.store.base.Store.exists] if you need
        to distinguish the two.

        ```python
        missing = object()
        obj = store.get(key, default=missing)
        if obj is missing:
            ...
        ```

    Args:
        key: Key associated with the object to retrieve.
        default: An optional value to be returned if an object
            associated with the key does not exist.

    Returns:
        Object or `default` if the object does not exist.

    Raises:
        SerializationError: If an exception is caught when deserializing
            the object associated with the key.
    """
    timer = Timer().start()

    cached = self.cache.get(key, _MISSING_OBJECT)
    if cached is not _MISSING_OBJECT:
        timer.stop()
        if self.metrics is not None:
            self.metrics.add_counter('store.get.cache_hits', key, 1)
            self.metrics.add_time('store.get', key, timer.elapsed_ms)

        logger.debug(
            'Store(%s): GET %s in %.3f ms (cached=True)',
            self._label,
            key,
            timer.elapsed_ms,
        )
        return self._from_cache(key, cached, timed=True)

    if self.metrics is not None:
        self.metrics.add_counter('store.get.cache_misses', key, 1)

    # Another thread may already be getting the object of the key, so
    # wait on that instead of getting it again.
    future, started = self.cache.start(key)
    if started:
        result = self._fetch(key, future)
    else:
        result = self._from_cache(key, future.result(), timed=True)

    if result is _MISSING_OBJECT:
        result = default

    timer.stop()
    if self.metrics is not None:
        self.metrics.add_time('store.get', key, timer.elapsed_ms)

    logger.debug(
        'Store(%s): GET %s in %.3f ms (cached=False)',
        self._label,
        key,
        timer.elapsed_ms,
    )
    return result

get_batch ¶

get_batch(
    keys: Sequence[KeyT], *, default: object | None = None
) -> list[Any | None]

Get the objects associated with the keys.

Cached objects are returned from the cache, and the remaining objects are retrieved with a single call to Connector.get_batch().

Parameters:

keys (Sequence[KeyT]) –

Sequence of keys associated with the objects to retrieve.
default (object | None, default: None ) –

An optional value to be returned for each object that does not exist.

Returns:

list[Any | None] –

List with the same order as keys containing each object or default if the object does not exist.

Raises:

SerializationError –

If an exception is caught when deserializing an object.

Source code in proxystore/store/base.py

def get_batch(
    self: Store[Connector[KeyT]],
    keys: Sequence[KeyT],
    *,
    default: object | None = None,
) -> list[Any | None]:
    """Get the objects associated with the keys.

    Cached objects are returned from the cache, and the remaining objects
    are retrieved with a single call to
    [`Connector.get_batch()`][proxystore.connectors.protocols.Connector.get_batch].

    Args:
        keys: Sequence of keys associated with the objects to retrieve.
        default: An optional value to be returned for each object
            that does not exist.

    Returns:
        List with the same order as `keys` containing each object or \
        `default` if the object does not exist.

    Raises:
        SerializationError: If an exception is caught when deserializing
            an object.
    """
    timer = Timer().start()
    results: list[Any] = [default] * len(keys)

    started, waiting, cached = self._start_batch(keys, results)
    misses = len(started) + len(waiting)

    ctime = dtime = 0.0
    if len(started) > 0:
        ctime, dtime = self._fetch_batch(keys, started, results)

    # With the bytes cache mode, cached and waited on values are
    # deserialized here.
    with Timer() as deserializer_timer:
        for i, future in waiting:
            result = self._from_cache(keys[i], future.result())
            if result is not _MISSING_OBJECT:
                results[i] = result
        for i in cached:
            results[i] = self._from_cache(keys[i], results[i])
    bytes_mode = self._cache_mode == 'bytes'
    if bytes_mode:
        dtime += deserializer_timer.elapsed_ms
    deserialized = len(started) > 0 or (
        bytes_mode and len(waiting) + len(cached) > 0
    )

    timer.stop()
    if self.metrics is not None:
        hits = len(keys) - misses
        self.metrics.add_counter('store.get_batch.cache_hits', keys, hits)
        self.metrics.add_counter(
            'store.get_batch.cache_misses',
            keys,
            misses,
        )
        if len(started) > 0:
            self.metrics.add_time('store.get_batch.connector', keys, ctime)
        if deserialized:
            self.metrics.add_time(
                'store.get_batch.deserialize',
                keys,
                dtime,
            )
        self.metrics.add_time('store.get_batch', keys, timer.elapsed_ms)

    logger.debug(
        'Store(%s): GET_BATCH (%s items, %s cached) in %.3f ms',
        self._label,
        len(keys),
        len(keys) - misses,
        timer.elapsed_ms,
    )
    return results

is_cached ¶

is_cached(key: KeyT) -> bool

Check if an object associated with the key is cached locally.

Parameters:

key (KeyT) –

Key potentially associated with stored object.

Returns:

bool –

If the object is cached.

Source code in proxystore/store/base.py

def is_cached(self: Store[Connector[KeyT]], key: KeyT) -> bool:
    """Check if an object associated with the key is cached locally.

    Args:
        key: Key potentially associated with stored object.

    Returns:
        If the object is cached.
    """
    return self.cache.exists(key)

proxy ¶

proxy(
    obj: NonProxiableT,
    *,
    evict: bool = ...,
    lifetime: Lifetime | None = ...,
    populate_target: bool | None = ...,
    skip_nonproxiable: Literal[True] = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> NonProxiableT

proxy(
    obj: T,
    *,
    evict: bool = ...,
    lifetime: Lifetime | None = ...,
    populate_target: bool | None = ...,
    skip_nonproxiable: bool = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> Proxy[T]

proxy(
    obj: T | NonProxiableT,
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> Proxy[T] | NonProxiableT

Create a proxy that will resolve to an object in the store.

Parameters:

obj (T | NonProxiableT) –

The object to place in store and return a proxy for.
evict (bool, default: False ) –

If the proxy should evict the object once resolved. Mutually exclusive with the lifetime parameter.
lifetime (Lifetime | None, default: None ) –

Attach the proxy to this lifetime. The object associated with the proxy will be evicted when the lifetime ends. Mutually exclusive with the evict parameter.
populate_target (bool | None, default: None ) –

Pass cache_defaults=True and target=obj to the Proxy constructor. I.e., return a proxy that (1) is already resolved, (2) can be used in isinstance checks without resolving, and (3) is hashable without resolving if obj is a hashable type. Note that the returned proxy will hold a reference to obj which will prevent garbage collecting obj. If None, defaults to the store-wide setting.
skip_nonproxiable (bool, default: True ) –

Return non-proxiable types (e.g., built-in constants like bool or None) directly. If False, a NonProxiableTypeError is raised instead.
connector_options (Mapping[str, Any] | None, default: None ) –

Additional keyword arguments to pass to Connector.put().

Returns:

Proxy[T] | NonProxiableT –

A proxy of the object unless obj is a non-proxiable type and skip_nonproxiable is True in which case obj is returned directly.

Raises:

NonProxiableTypeError –

If obj is a non-proxiable type and skip_nonproxiable=False.
ValueError –

If evict is True and lifetime is not None because these parameters are mutually exclusive.

Source code in proxystore/store/base.py

def proxy(
    self,
    obj: T | NonProxiableT,
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> Proxy[T] | NonProxiableT:
    """Create a proxy that will resolve to an object in the store.

    Args:
        obj: The object to place in store and return a proxy for.
        evict: If the proxy should evict the object once resolved.
            Mutually exclusive with the `lifetime` parameter.
        lifetime: Attach the proxy to this lifetime. The object associated
            with the proxy will be evicted when the lifetime ends.
            Mutually exclusive with the `evict` parameter.
        populate_target: Pass `cache_defaults=True` and `target=obj` to
            the [`Proxy`][proxystore.proxy.Proxy] constructor. I.e.,
            return a proxy that (1) is already resolved, (2) can be used
            in [`isinstance`][isinstance] checks without resolving, and (3)
            is hashable without resolving if `obj` is a hashable type.
            Note that the returned proxy will hold a reference to `obj`
            which will prevent garbage collecting `obj`. If `None`,
            defaults to the store-wide setting.
        skip_nonproxiable: Return non-proxiable types (e.g., built-in
            constants like `bool` or `None`) directly. If `False`, a
            [`NonProxiableTypeError`][proxystore.store.exceptions.NonProxiableTypeError]
            is raised instead.
        connector_options: Additional keyword arguments to pass to
            [`Connector.put()`][proxystore.connectors.protocols.Connector.put].

    Returns:
        A proxy of the object unless `obj` is a non-proxiable type \
        and `#!python skip_nonproxiable is True` in which case `obj` is \
        returned directly.

    Raises:
        NonProxiableTypeError: If `obj` is a non-proxiable type and
            `#!python skip_nonproxiable=False`.
        ValueError: If `evict` is `True` and `lifetime` is not `None`
            because these parameters are mutually exclusive.
    """
    if evict and lifetime is not None:
        raise ValueError(
            'The evict and lifetime parameters are mutually exclusive. '
            'Only set one of evict or lifetime.',
        )

    if isinstance(obj, _NON_PROXIABLE_TYPES):
        if skip_nonproxiable:
            # MyPy raises the following error which is not correct:
            #     Incompatible return value type (got "Optional[bool]",
            #     expected "Optional[Proxy[T]]")  [return-value]
            return obj  # type: ignore[return-value]
        raise NonProxiableTypeError(
            f'Object of {type(obj)} is not proxiable.',
        )

    with Timer() as timer:
        key = self.put(
            obj,
            connector_options=connector_options,
        )
        factory: StoreFactory[ConnectorT, T] = StoreFactory(
            key,
            store_config=self.config(),
            evict=evict,
        )
        populate_target = (
            self._populate_target
            if populate_target is None
            else populate_target
        )
        if populate_target:
            # If obj were None, we would have escaped early when
            # checking _NON_PROXIABLE_TYPES.
            assert obj is not None
            proxy = Proxy(factory, cache_defaults=True, target=obj)
        else:
            proxy = Proxy(factory)

        if lifetime is not None:
            lifetime.add_proxy(proxy)

    if self.metrics is not None:
        self.metrics.add_time('store.proxy', key, timer.elapsed_ms)

    logger.debug(
        'Store(%s): PROXY %s in %.3f ms',
        self._label,
        key,
        timer.elapsed_ms,
    )
    return proxy

proxy_batch ¶

proxy_batch(
    objs: Sequence[NonProxiableT],
    *,
    evict: bool = ...,
    lifetime: Lifetime | None = ...,
    populate_target: bool | None = ...,
    skip_nonproxiable: Literal[True] = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> list[NonProxiableT]

proxy_batch(
    objs: Sequence[T],
    *,
    evict: bool = ...,
    lifetime: Lifetime | None = ...,
    populate_target: bool | None = ...,
    skip_nonproxiable: bool = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> list[Proxy[T]]

proxy_batch(
    objs: Sequence[T | NonProxiableT],
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> list[Proxy[T] | NonProxiableT]

Create proxies that will resolve to an object in the store.

Parameters:

objs (Sequence[T | NonProxiableT]) –

The objects to place in store and return a proxies for.
evict (bool, default: False ) –

If a proxy should evict its object once resolved. Mutually exclusive with the lifetime parameter.
lifetime (Lifetime | None, default: None ) –

Attach the proxies to this lifetime. The objects associated with each proxy will be evicted when the lifetime ends. Mutually exclusive with the evict parameter.
populate_target (bool | None, default: None ) –

Pass cache_defaults=True and target=obj to the Proxy constructor. I.e., return a proxy that (1) is already resolved, (2) can be used in isinstance checks without resolving, and (3) is hashable without resolving if obj is a hashable type. If None, defaults to the store-wide setting.
skip_nonproxiable (bool, default: True ) –

Return non-proxiable types (e.g., built-in constants like bool or None) directly. If False, a NonProxiableTypeError is raised instead.
connector_options (Mapping[str, Any] | None, default: None ) –

Additional keyword arguments to pass to Connector.put_batch().

Returns:

list[Proxy[T] | NonProxiableT] –

A list of proxies of each object or the object itself if said object is not proxiable and skip_nonproxiable is True.

Raises:

NonProxiableTypeError –

If obj is a non-proxiable type and skip_nonproxiable=False.
ValueError –

If evict is True and lifetime is not None because these parameters are mutually exclusive.

Source code in proxystore/store/base.py

def proxy_batch(  # type: ignore[misc]
    self,
    objs: Sequence[T | NonProxiableT],
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> list[Proxy[T] | NonProxiableT]:
    """Create proxies that will resolve to an object in the store.

    Args:
        objs: The objects to place in store and return a proxies for.
        evict: If a proxy should evict its object once resolved.
            Mutually exclusive with the `lifetime` parameter.
        lifetime: Attach the proxies to this lifetime. The objects
            associated with each proxy will be evicted when the lifetime
            ends. Mutually exclusive with the `evict` parameter.
        populate_target: Pass `cache_defaults=True` and `target=obj` to
            the [`Proxy`][proxystore.proxy.Proxy] constructor. I.e.,
            return a proxy that (1) is already resolved, (2) can be used
            in [`isinstance`][isinstance] checks without resolving, and (3)
            is hashable without resolving if `obj` is a hashable type.
            If `None`, defaults to the store-wide setting.
        skip_nonproxiable: Return non-proxiable types (e.g., built-in
            constants like `bool` or `None`) directly. If `False`, a
            [`NonProxiableTypeError`][proxystore.store.exceptions.NonProxiableTypeError]
            is raised instead.
        connector_options: Additional keyword arguments to pass to
            [`Connector.put_batch()`][proxystore.connectors.protocols.Connector.put_batch].

    Returns:
        A list of proxies of each object or the object itself if said \
        object is not proxiable and `#!python skip_nonproxiable is True`.

    Raises:
        NonProxiableTypeError: If `obj` is a non-proxiable type and
            `#!python skip_nonproxiable=False`.
        ValueError: If `evict` is `True` and `lifetime` is not `None`
            because these parameters are mutually exclusive.
    """
    if evict and lifetime is not None:
        raise ValueError(
            'The evict and lifetime parameters are mutually exclusive. '
            'Only set one of evict or lifetime.',
        )

    with Timer() as timer:
        # Find if there are non-proxiable types and if that's okay
        non_proxiable: list[tuple[int, Any]] = []
        for i, obj in enumerate(objs):
            if isinstance(obj, _NON_PROXIABLE_TYPES):
                non_proxiable.append((i, obj))

        if len(non_proxiable) > 0 and not skip_nonproxiable:
            raise NonProxiableTypeError(
                f'Input sequence contains {len(non_proxiable)} '
                'objects that are not proxiable.',
            )

        # Pop non-proxiable types so we can batch proxy the proxiable ones
        non_proxiable_indicies = [i for i, _ in non_proxiable]
        proxiable_objs = [
            obj
            for i, obj in enumerate(objs)
            if i not in non_proxiable_indicies
        ]

        keys = self.put_batch(
            proxiable_objs,
            connector_options=connector_options,
        )
        factories: list[StoreFactory[ConnectorT, T]] = [
            StoreFactory(
                key,
                store_config=self.config(),
                evict=evict,
            )
            for key in keys
        ]

        populate_target = (
            self._populate_target
            if populate_target is None
            else populate_target
        )

        proxies: list[Proxy[T]] = []
        for factory, obj in zip(factories, proxiable_objs, strict=True):
            if populate_target:
                proxy = Proxy(factory, cache_defaults=True, target=obj)
            else:
                proxy = Proxy(factory)
            proxies.append(proxy)

        if lifetime is not None:
            lifetime.add_proxy(*proxies)

        # Put non-proxiable objects back in their original positions.
        # The indices of non_proxiable must still be sorted
        for original_index, original_object in non_proxiable:
            proxies.insert(original_index, original_object)

    if self.metrics is not None:
        self.metrics.add_time('store.proxy_batch', keys, timer.elapsed_ms)

    logger.debug(
        'Store(%s): PROXY_BATCH (%s items) in %.3f ms',
        self._label,
        len(proxies),
        timer.elapsed_ms,
    )
    return cast(list[Proxy[T] | NonProxiableT], proxies)

proxy_from_key ¶

proxy_from_key(
    key: KeyT,
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
) -> Proxy[T]

Create a proxy that will resolve to an object already in the store.

Parameters:

key (KeyT) –

The key associated with an object already in the store.
evict (bool, default: False ) –

If the proxy should evict the object once resolved. Mutually exclusive with the lifetime parameter.
lifetime (Lifetime | None, default: None ) –

Attach the proxy to this lifetime. The object associated with the proxy will be evicted when the lifetime ends. Mutually exclusive with the evict parameter.

Returns:

Proxy[T] –

A proxy of the object.

Raises:

ValueError –

If evict is True and lifetime is not None because these parameters are mutually exclusive.

Source code in proxystore/store/base.py

def proxy_from_key(
    self: Store[Connector[KeyT]],
    key: KeyT,
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
) -> Proxy[T]:
    """Create a proxy that will resolve to an object already in the store.

    Args:
        key: The key associated with an object already in the store.
        evict: If the proxy should evict the object once resolved.
            Mutually exclusive with the `lifetime` parameter.
        lifetime: Attach the proxy to this lifetime. The object associated
            with the proxy will be evicted when the lifetime ends.
            Mutually exclusive with the `evict` parameter.

    Returns:
        A proxy of the object.

    Raises:
        ValueError: If `evict` is `True` and `lifetime` is not `None`
            because these parameters are mutually exclusive.
    """
    if evict and lifetime is not None:
        raise ValueError(
            'The evict and lifetime parameters are mutually exclusive. '
            'Only set one of evict or lifetime.',
        )

    factory: StoreFactory[ConnectorT, T] = StoreFactory(
        key,
        store_config=self.config(),
        evict=evict,
    )
    proxy = Proxy(factory)

    logger.debug('Store(%s): PROXY_FROM_KEY %s', self._label, key)

    if lifetime is not None:
        lifetime.add_proxy(proxy)

    return proxy

locked_proxy ¶

locked_proxy(
    obj: NonProxiableT,
    *,
    evict: bool = ...,
    lifetime: Lifetime | None = ...,
    populate_target: bool | None = ...,
    skip_nonproxiable: Literal[True] = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> NonProxiableT

locked_proxy(
    obj: T,
    *,
    evict: bool = ...,
    lifetime: Lifetime | None = ...,
    populate_target: bool | None = ...,
    skip_nonproxiable: bool = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> ProxyLocker[T]

locked_proxy(
    obj: T | NonProxiableT,
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> ProxyLocker[T] | NonProxiableT

Proxy an object and return ProxyLocker.

Parameters:

obj (T | NonProxiableT) –

The object to place in store and return a proxy for.
evict (bool, default: False ) –

If the proxy should evict the object once resolved. Mutually exclusive with the lifetime parameter.
lifetime (Lifetime | None, default: None ) –

Attach the proxy to this lifetime. The object associated with the proxy will be evicted when the lifetime ends. Mutually exclusive with the evict parameter.
populate_target (bool | None, default: None ) –

Pass cache_defaults=True and target=obj to the Proxy constructor. I.e., return a proxy that (1) is already resolved, (2) can be used in isinstance checks without resolving, and (3) is hashable without resolving if obj is a hashable type. If None, defaults to the store-wide setting.
skip_nonproxiable (bool, default: True ) –

Return non-proxiable types (e.g., built-in constants like bool or None) directly. If False, a NonProxiableTypeError is raised instead.
connector_options (Mapping[str, Any] | None, default: None ) –

Additional keyword arguments to pass to Connector.put().

Returns:

ProxyLocker[T] | NonProxiableT –

A proxy wrapped in a ProxyLocker unless obj is a non-proxiable type and skip_nonproxiable is True in which case obj is returned directly.

Raises:

NonProxiableTypeError –

If obj is a non-proxiable type and skip_nonproxiable=False.
ValueError –

If evict is True and lifetime is not None because these parameters are mutually exclusive.

Source code in proxystore/store/base.py

def locked_proxy(
    self,
    obj: T | NonProxiableT,
    *,
    evict: bool = False,
    lifetime: Lifetime | None = None,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> ProxyLocker[T] | NonProxiableT:
    """Proxy an object and return [`ProxyLocker`][proxystore.proxy.ProxyLocker].

    Args:
        obj: The object to place in store and return a proxy for.
        evict: If the proxy should evict the object once resolved.
            Mutually exclusive with the `lifetime` parameter.
        lifetime: Attach the proxy to this lifetime. The object associated
            with the proxy will be evicted when the lifetime ends.
            Mutually exclusive with the `evict` parameter.
        populate_target: Pass `cache_defaults=True` and `target=obj` to
            the [`Proxy`][proxystore.proxy.Proxy] constructor. I.e.,
            return a proxy that (1) is already resolved, (2) can be used
            in [`isinstance`][isinstance] checks without resolving, and (3)
            is hashable without resolving if `obj` is a hashable type.
            If `None`, defaults to the store-wide setting.
        skip_nonproxiable: Return non-proxiable types (e.g., built-in
            constants like `bool` or `None`) directly. If `False`, a
            [`NonProxiableTypeError`][proxystore.store.exceptions.NonProxiableTypeError]
            is raised instead.
        connector_options: Additional keyword arguments to pass to
            [`Connector.put()`][proxystore.connectors.protocols.Connector.put].

    Returns:
        A proxy wrapped in a \
        [`ProxyLocker`][proxystore.proxy.ProxyLocker] unless `obj` is a \
        non-proxiable type and `#!python skip_nonproxiable is True` in which \
        case `obj` is returned directly.

    Raises:
        NonProxiableTypeError: If `obj` is a non-proxiable type and
            `#!python skip_nonproxiable=False`.
        ValueError: If `evict` is `True` and `lifetime` is not `None`
            because these parameters are mutually exclusive.
    """  # noqa: E501
    possible_proxy = self.proxy(
        obj,
        evict=evict,
        lifetime=lifetime,
        populate_target=populate_target,
        skip_nonproxiable=skip_nonproxiable,
        connector_options=connector_options,
    )

    if isinstance(possible_proxy, Proxy):
        return ProxyLocker(possible_proxy)
    return possible_proxy

owned_proxy ¶

owned_proxy(
    obj: NonProxiableT,
    *,
    populate_target: bool | None = ...,
    skip_nonproxiable: Literal[True] = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> NonProxiableT

owned_proxy(
    obj: T,
    *,
    populate_target: bool | None = ...,
    skip_nonproxiable: bool = ...,
    connector_options: Mapping[str, Any] | None = ...,
) -> OwnedProxy[T]

owned_proxy(
    obj: T | NonProxiableT,
    *,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> OwnedProxy[T] | NonProxiableT

Create a proxy that will enforce ownership rules over the object.

An OwnedProxy will auto-evict the object once it goes out of scope. This proxy type can also be borrowed.

Parameters:

obj (T | NonProxiableT) –

The object to place in store and return a proxy for.
populate_target (bool | None, default: None ) –

Pass cache_defaults=True and target=obj to the Proxy constructor. I.e., return a proxy that (1) is already resolved, (2) can be used in isinstance checks without resolving, and (3) is hashable without resolving if obj is a hashable type. If None, defaults to the store-wide setting.
skip_nonproxiable (bool, default: True ) –

Return non-proxiable types (e.g., built-in constants like bool or None) directly. If False, a NonProxiableTypeError is raised instead.
connector_options (Mapping[str, Any] | None, default: None ) –

Additional keyword arguments to pass to Connector.put().

Returns:

OwnedProxy[T] | NonProxiableT –

A proxy of the object unless obj is a non-proxiable type and skip_nonproxiable is True in which case obj is returned directly.

Raises:

NonProxiableTypeError –

If obj is a non-proxiable type and skip_nonproxiable=False.

Source code in proxystore/store/base.py

def owned_proxy(
    self,
    obj: T | NonProxiableT,
    *,
    populate_target: bool | None = None,
    skip_nonproxiable: bool = True,
    connector_options: Mapping[str, Any] | None = None,
) -> OwnedProxy[T] | NonProxiableT:
    """Create a proxy that will enforce ownership rules over the object.

    An [`OwnedProxy`][proxystore.store.ref.OwnedProxy] will auto-evict
    the object once it goes out of scope. This proxy type can also
    be borrowed.

    Args:
        obj: The object to place in store and return a proxy for.
        populate_target: Pass `cache_defaults=True` and `target=obj` to
            the [`Proxy`][proxystore.proxy.Proxy] constructor. I.e.,
            return a proxy that (1) is already resolved, (2) can be used
            in [`isinstance`][isinstance] checks without resolving, and (3)
            is hashable without resolving if `obj` is a hashable type.
            If `None`, defaults to the store-wide setting.
        skip_nonproxiable: Return non-proxiable types (e.g., built-in
            constants like `bool` or `None`) directly. If `False`, a
            [`NonProxiableTypeError`][proxystore.store.exceptions.NonProxiableTypeError]
            is raised instead.
        connector_options: Additional keyword arguments to pass to
            [`Connector.put()`][proxystore.connectors.protocols.Connector.put].

    Returns:
        A proxy of the object unless `obj` is a non-proxiable type \
        and `#!python skip_nonproxiable is True` in which case `obj` is \
        returned directly.

    Raises:
        NonProxiableTypeError: If `obj` is a non-proxiable type and
            `#!python skip_nonproxiable=False`.
    """
    possible_proxy = self.proxy(
        obj,
        evict=False,
        populate_target=populate_target,
        skip_nonproxiable=skip_nonproxiable,
        connector_options=connector_options,
    )

    if isinstance(possible_proxy, Proxy):
        populate_target = (
            self._populate_target
            if populate_target is None
            else populate_target
        )
        return into_owned(possible_proxy, populate_target=populate_target)
    return possible_proxy

put ¶

put(
    obj: Any,
    *,
    lifetime: Lifetime | None = None,
    connector_options: Mapping[str, Any] | None = None,
) -> KeyT

Put an object in the store.

Parameters:

obj (Any) –

Object to put in the store.
lifetime (Lifetime | None, default: None ) –

Attach the key to this lifetime. The object associated with the key will be evicted when the lifetime ends.
connector_options (Mapping[str, Any] | None, default: None ) –

Additional keyword arguments to pass to Connector.put().

Returns:

KeyT –

A key which can be used to retrieve the object.

Raises:

TypeError –

If the output of the serializer is not bytes.

Source code in proxystore/store/base.py

def put(
    self: Store[Connector[KeyT]],
    obj: Any,
    *,
    lifetime: Lifetime | None = None,
    connector_options: Mapping[str, Any] | None = None,
) -> KeyT:
    """Put an object in the store.

    Args:
        obj: Object to put in the store.
        lifetime: Attach the key to this lifetime. The object associated
            with the key will be evicted when the lifetime ends.
        connector_options: Additional keyword arguments to pass to
            [`Connector.put()`][proxystore.connectors.protocols.Connector.put].

    Returns:
        A key which can be used to retrieve the object.

    Raises:
        TypeError: If the output of the serializer is not bytes.
    """
    timer = Timer().start()

    with Timer() as serialize_timer:
        obj = self.serializer(obj)

    if not is_bytes_like(obj):
        raise TypeError('Serializer must return a bytes-like object.')

    with Timer() as connector_timer:
        key = self.connector.put(obj, **(connector_options or {}))

    if lifetime is not None:
        lifetime.add_key(key, store=self)

    timer.stop()
    if self.metrics is not None:
        ctime = connector_timer.elapsed_ms
        stime = serialize_timer.elapsed_ms
        self.metrics.add_attribute('store.put.object_size', key, len(obj))
        self.metrics.add_time('store.put.serialize', key, stime)
        self.metrics.add_time('store.put.connector', key, ctime)
        self.metrics.add_time('store.put', key, timer.elapsed_ms)

    logger.debug(
        'Store(%s): PUT %s in %.3f ms',
        self._label,
        key,
        timer.elapsed_ms,
    )
    return key

put_batch ¶

put_batch(
    objs: Sequence[Any],
    *,
    lifetime: Lifetime | None = None,
    connector_options: Mapping[str, Any] | None = None,
) -> list[KeyT]

Put multiple objects in the store.

Parameters:

objs (Sequence[Any]) –

Sequence of objects to put in the store.
lifetime (Lifetime | None, default: None ) –

Attach the keys to this lifetime. The objects associated with each key will be evicted when the lifetime ends.
connector_options (Mapping[str, Any] | None, default: None ) –

Additional keyword arguments to pass to Connector.put_batch().

Returns:

list[KeyT] –

A list of keys which can be used to retrieve the objects.

Raises:

TypeError –

If the output of the serializer is not bytes.

Source code in proxystore/store/base.py

def put_batch(
    self: Store[Connector[KeyT]],
    objs: Sequence[Any],
    *,
    lifetime: Lifetime | None = None,
    connector_options: Mapping[str, Any] | None = None,
) -> list[KeyT]:
    """Put multiple objects in the store.

    Args:
        objs: Sequence of objects to put in the store.
        lifetime: Attach the keys to this lifetime. The objects associated
            with each key will be evicted when the lifetime ends.
        connector_options: Additional keyword arguments to pass to
            [`Connector.put_batch()`][proxystore.connectors.protocols.Connector.put_batch].

    Returns:
        A list of keys which can be used to retrieve the objects.

    Raises:
        TypeError: If the output of the serializer is not bytes.
    """
    timer = Timer().start()

    def _serialize(obj: Any) -> BytesLike:
        obj = self.serializer(obj)

        if not is_bytes_like(obj):
            raise TypeError('Serializer must return a bytes-like object.')

        return obj

    with Timer() as serialize_timer:
        _objs = list(map(_serialize, objs))

    with Timer() as connector_timer:
        keys = self.connector.put_batch(
            _objs,
            **(connector_options or {}),
        )

    if lifetime is not None:
        lifetime.add_key(*keys, store=self)

    timer.stop()
    if self.metrics is not None:
        ctime = connector_timer.elapsed_ms
        stime = serialize_timer.elapsed_ms
        sizes = sum(len(obj) for obj in _objs)
        self.metrics.add_attribute(
            'store.put_batch.object_sizes',
            keys,
            sizes,
        )
        self.metrics.add_time('store.put_batch.serialize', keys, stime)
        self.metrics.add_time('store.put_batch.connector', keys, ctime)
        self.metrics.add_time('store.put_batch', keys, timer.elapsed_ms)

    logger.debug(
        'Store(%s): PUT_BATCH (%s items) in %.3f ms',
        self._label,
        len(keys),
        timer.elapsed_ms,
    )
    return keys

StoreConfig ¶

Bases: BaseModel

Store configuration.

Tip

See the Store parameters for more information about each configuration option.

Attributes:

id (str | None) –

Unique ID of the store. Stores initialized from configurations with the same ID are considered to be the same store, so a proxy resolved in a process where a store with the same ID exists will use that store. If None, a new ID is generated when a store is initialized from the configuration. The ID is not included when writing the configuration to a TOML file.
name (str | None) –

Optional store name.
connector (ConnectorConfig) –

Connector configuration.
serializer (SerializerT | None) –

Optional serializer.
deserializer (DeserializerT | None) –

Optional deserializer.
cache_size_bytes (int) –

Maximum cache size in bytes.
cache_mode (CacheModeT) –

Cache deserialized objects or serialized data.
metrics (bool) –

Enable recording operation metrics.
populate_target (bool) –

Set the default value for the populate_target parameter of proxy methods.

from_toml `classmethod` ¶

from_toml(filepath: str | Path) -> Self

Create a configuration file from a TOML file.

Example

See write_toml().

Parameters:

filepath (str | Path) –

Path to TOML file to load.

Source code in proxystore/store/config.py

@classmethod
def from_toml(cls, filepath: str | pathlib.Path) -> Self:
    """Create a configuration file from a TOML file.

    Example:
        See
        [`write_toml()`][proxystore.store.config.StoreConfig.write_toml].

    Args:
        filepath: Path to TOML file to load.
    """
    with open(filepath, 'rb') as f:
        return load(cls, f)

write_toml ¶

write_toml(filepath: str | Path) -> None

Write a configuration to a TOML file.

Example

from proxystore.store.config import ConnectorConfig
from proxystore.store.config import StoreConfig

config = StoreConfig(
    name='example',
    connector=ConnectorConfig(
        kind='file',
        options={'store_dir': '/tmp/proxystore-cache'},
    ),
)

config.write_toml('config.toml')

The resulting TOML file contains the full configuration, including default options but excluding the id, and can be loaded again using StoreConfig.from_toml('config.toml').

config.toml

name = "example"
cache_size_bytes = 268435456
cache_mode = "objects"
metrics = false
populate_target = true

[connector]
kind = "file"

[connector.options]
store_dir = "/tmp/proxystore-cache"

Parameters:

filepath (str | Path) –

Path to TOML file to write.

Source code in proxystore/store/config.py

def write_toml(self, filepath: str | pathlib.Path) -> None:
    """Write a configuration to a TOML file.

    Example:
        ```python
        from proxystore.store.config import ConnectorConfig
        from proxystore.store.config import StoreConfig

        config = StoreConfig(
            name='example',
            connector=ConnectorConfig(
                kind='file',
                options={'store_dir': '/tmp/proxystore-cache'},
            ),
        )

        config.write_toml('config.toml')
        ```
        The resulting TOML file contains the full configuration,
        including default options but excluding the
        [`id`][proxystore.store.config.StoreConfig], and can be loaded
        again using `#!python StoreConfig.from_toml('config.toml')`.
        ```toml title="config.toml"
        name = "example"
        cache_size_bytes = 268435456
        cache_mode = "objects"
        metrics = false
        populate_target = true

        [connector]
        kind = "file"

        [connector.options]
        store_dir = "/tmp/proxystore-cache"
        ```

    Args:
        filepath: Path to TOML file to write.
    """
    filepath = pathlib.Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)
    # The ID is excluded so that stores initialized from the same file
    # in different processes do not share an ID unintentionally.
    with open(filepath, 'wb') as f:
        dump(self.model_copy(update={'id': None}), f)

StoreFactory ¶

StoreFactory(
    key: ConnectorKeyT,
    store_config: StoreConfig,
    *,
    evict: bool = False,
)

Bases: Generic[ConnectorT, T]

Factory that resolves an object from a store.

Adds support for asynchronously retrieving objects from a Store instance.

The factory takes the store_config parameter that is used to reinitialize the store if the factory is sent to a remote process where the store has not already been initialized.

Calls to the factory from threads while another call (or resolve_async()) is resolving the object wait for that result, so the object is resolved once and every thread gets the same object.

Parameters:

key (ConnectorKeyT) –

Key corresponding to object in store.
store_config (StoreConfig) –

Store configuration used to reinitialize the store if needed.
evict (bool, default: False ) –

If True, evict the object from the store once resolve() is called.

Source code in proxystore/store/factory.py

def __init__(
    self,
    key: ConnectorKeyT,
    store_config: StoreConfig,
    *,
    evict: bool = False,
) -> None:
    # The key is typed Any because the type of the store's connector,
    # and so the type of its keys, is not known from a proxy.
    self.key: Any = key
    self.store_config = store_config
    self.evict = evict

    # Not included when a factory is serialized because it is specific
    # to this instance of the factory.
    self._resolve: _Resolve | None = None

get_store ¶

get_store() -> Store[ConnectorT]

Get store and reinitialize if necessary.

Raises:

StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/factory.py

def get_store(self) -> Store[ConnectorT]:
    """Get store and reinitialize if necessary.

    Raises:
        StoreClosedError: If the owner of the store closed it and
            removed its objects in this process.
    """
    return registry.get_or_create(self.store_config)

resolve ¶

resolve() -> T

Get object associated with key from store.

Raises:

ProxyResolveMissingKeyError –

If the key associated with this factory does not exist in the store.
StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/factory.py

def resolve(self) -> T:
    """Get object associated with key from store.

    Raises:
        ProxyResolveMissingKeyError: If the key associated with this
            factory does not exist in the store.
        StoreClosedError: If the owner of the store closed it and
            removed its objects in this process.
    """
    with Timer() as timer:
        store = self.get_store()
        obj = store.get(self.key, default=_MISSING_OBJECT)

        if obj is _MISSING_OBJECT:
            raise ProxyResolveMissingKeyError(
                self.key,
                type(store),
                store.name,
                store.id,
            )

        if self.evict:
            store.evict(self.key)

    if store.metrics is not None:
        total_time = timer.elapsed_ms
        store.metrics.add_time('factory.resolve', self.key, total_time)

    return cast(T, obj)

resolve_async ¶

resolve_async() -> None

Asynchronously get object associated with key from store.

Nothing happens if the object is already being resolved.

Source code in proxystore/store/factory.py

def resolve_async(self) -> None:
    """Asynchronously get object associated with key from store.

    Nothing happens if the object is already being resolved.
    """
    with _resolve_lock:
        current = self._resolve
        if current is not None and current.generation == _fork_generation:
            return
        logger.debug('Starting asynchronous resolve of %s', self.key)
        resolve = _Resolve(thread=None)
        resolve.future = _get_default_pool().submit(
            self._resolve_in_pool,
            resolve,
        )
        self._resolve = resolve

get_store ¶

get_store(proxy: Proxy[T]) -> Store[Any]

Get the store which created a proxy.

The store is initialized and registered from the configuration contained in the proxy's factory if the store does not already exist in this process.

Parameters:

proxy (Proxy[T]) –

Proxy instance created by a Store.

Returns:

Store[Any] –

Store which created the proxy.

Raises:

ProxyStoreFactoryError –

If the proxy does not contain a factory of type StoreFactory.
StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/__init__.py

def get_store(proxy: Proxy[T]) -> Store[Any]:
    """Get the store which created a proxy.

    The store is initialized and registered from the configuration
    contained in the proxy's factory if the store does not already exist
    in this process.

    Args:
        proxy: [`Proxy`][proxystore.proxy.Proxy] instance created by a
            [`Store`][proxystore.store.base.Store].

    Returns:
        [`Store`][proxystore.store.base.Store] which created the proxy.

    Raises:
        ProxyStoreFactoryError: If the proxy does not contain a factory of
            type [`StoreFactory`][proxystore.store.factory.StoreFactory].
        StoreClosedError: If the owner of the store closed it and removed
            its objects in this process.
    """
    if type(proxy) is str:
        raise TypeError(
            'get_store() takes a proxy, not a store name. Looking up stores '
            'by name was removed in ProxyStore v2. Keep a reference to the '
            'store, or pass store.config() to other processes and use '
            'get_or_create_store().',
        )
    factory = get_factory(proxy)
    if isinstance(factory, StoreFactory):
        return factory.get_store()
    raise ProxyStoreFactoryError(
        'The proxy must contain a factory with type '
        f'{StoreFactory.__name__}. {type(factory).__name__} '
        'is not supported.',
    )

get_or_create_store ¶

get_or_create_store(
    store_config: StoreConfig,
) -> Store[Any]

Get a registered store or initialize a new instance from the config.

Note

A new store is not the owner of the objects stored by its connector, so closing the store does not clear the connector.

Parameters:

store_config (StoreConfig) –

Store configuration. If a store with the same id is registered, that store is returned. Otherwise, a new store is initialized (and registered) from the configuration.

Returns:

Store[Any] –

Store instance.

Raises:

StoreClosedError –

If the owner of the store closed it and removed its objects in this process.

Source code in proxystore/store/__init__.py

@ignore_register_kwarg
def get_or_create_store(store_config: StoreConfig) -> Store[Any]:
    """Get a registered store or initialize a new instance from the config.

    Note:
        A new store is not the
        [`owner`][proxystore.store.base.Store.owner] of the objects stored
        by its connector, so closing the store does not clear the connector.

    Args:
        store_config: Store configuration. If a store with the same
            [`id`][proxystore.store.config.StoreConfig] is registered, that
            store is returned. Otherwise, a new store is initialized (and
            registered) from the configuration.

    Returns:
        [`Store`][proxystore.store.base.Store] instance.

    Raises:
        StoreClosedError: If the owner of the store closed it and removed
            its objects in this process.
    """
    return _registry.get_or_create(store_config)

proxystore.store¶

Store ¶

connector property ¶

cache property ¶

id property ¶

owner property ¶

name property ¶

metrics property ¶

serializer property ¶

deserializer property ¶

close ¶

config ¶

from_config classmethod ¶

future ¶

evict ¶

exists ¶

get ¶

get_batch ¶

is_cached ¶

proxy ¶

proxy_batch ¶

proxy_from_key ¶

locked_proxy ¶

owned_proxy ¶

put ¶

put_batch ¶

StoreConfig ¶

from_toml classmethod ¶

write_toml ¶

StoreFactory ¶

get_store ¶

resolve ¶

resolve_async ¶

get_store ¶

get_or_create_store ¶

connector `property` ¶

cache `property` ¶

id `property` ¶

owner `property` ¶

name `property` ¶

metrics `property` ¶

serializer `property` ¶

deserializer `property` ¶

from_config `classmethod` ¶

from_toml `classmethod` ¶