Skip to content

proxystore.endpoint.storage

proxystore/endpoint/storage.py

Blob storage interface for endpoints.

Storage

Bases: Protocol

Endpoint storage protocol for blobs.

evict() async 

evict(key: str) -> None

Evict a blob from storage.

Parameters:

  • key (str) –

    Key associated with blob to evict.

Source code in proxystore/endpoint/storage.py 
20
21
22
23
24
25
26
async def evict(self, key: str) -> None:
    """Evict a blob from storage.

    Args:
        key: Key associated with blob to evict.
    """
    ...

exists() async 

exists(key: str) -> bool

Check if a blob exists in the storage.

Parameters:

  • key (str) –

    Key associated with the blob to check.

Returns:

  • bool –

    If a blob associated with the key exists.

Source code in proxystore/endpoint/storage.py 
28
29
30
31
32
33
34
35
36
37
async def exists(self, key: str) -> bool:
    """Check if a blob exists in the storage.

    Args:
        key: Key associated with the blob to check.

    Returns:
        If a blob associated with the key exists.
    """
    ...

get() async 

get(key: str, default: bytes | None = None) -> bytes | None

Get a blob from storage.

Parameters:

  • key (str) –

    Key associated with the blob to get.

  • default (bytes | None, default: None ) –

    Default return value if the blob does not exist.

Returns:

  • bytes | None –

    The blob associated with the key or the value of default.

Source code in proxystore/endpoint/storage.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
async def get(
    self,
    key: str,
    default: bytes | None = None,
) -> bytes | None:
    """Get a blob from storage.

    Args:
        key: Key associated with the blob to get.
        default: Default return value if the blob does not exist.

    Returns:
        The blob associated with the key or the value of `default`.
    """
    ...

set() async 

set(key: str, blob: bytes) -> None

Store the blob associated with a key.

Parameters:

  • key (str) –

    Key that will be used to retrieve the blob.

  • blob (bytes) –

    Blob to store.

Raises:

Source code in proxystore/endpoint/storage.py 
55
56
57
58
59
60
61
62
63
64
65
66
async def set(self, key: str, blob: bytes) -> None:
    """Store the blob associated with a key.

    Args:
        key: Key that will be used to retrieve the blob.
        blob: Blob to store.

    Raises:
        ObjectSizeExceededError: If the max object size is configured and
            the data exceeds that size.
    """
    ...

close() async 

close() -> None

Close the storage.

Source code in proxystore/endpoint/storage.py 
68
69
70
async def close(self) -> None:
    """Close the storage."""
    ...

DictStorage 

DictStorage(
    *, max_object_size: int | None = MAX_OBJECT_SIZE_DEFAULT
)

Simple dictionary-based storage for blobs.

Parameters:

  • max_object_size (int | None, default: MAX_OBJECT_SIZE_DEFAULT ) –

    Optional max size in bytes for any single object stored by the endpoint. If exceeded, an error is raised.

Source code in proxystore/endpoint/storage.py 
81
82
83
84
85
86
87
def __init__(
    self,
    *,
    max_object_size: int | None = MAX_OBJECT_SIZE_DEFAULT,
) -> None:
    self._data: dict[str, bytes] = {}
    self._max_object_size = max_object_size

evict() async 

evict(key: str) -> None

Evict a blob from storage.

Parameters:

  • key (str) –

    Key associated with blob to evict.

Source code in proxystore/endpoint/storage.py 
89
90
91
92
93
94
95
async def evict(self, key: str) -> None:
    """Evict a blob from storage.

    Args:
        key: Key associated with blob to evict.
    """
    self._data.pop(key, None)

exists() async 

exists(key: str) -> bool

Check if a blob exists in the storage.

Parameters:

  • key (str) –

    Key associated with the blob to check.

Returns:

  • bool –

    If a blob associated with the key exists.

Source code in proxystore/endpoint/storage.py 
 97
 98
 99
100
101
102
103
104
105
106
async def exists(self, key: str) -> bool:
    """Check if a blob exists in the storage.

    Args:
        key: Key associated with the blob to check.

    Returns:
        If a blob associated with the key exists.
    """
    return key in self._data

get() async 

get(key: str, default: bytes | None = None) -> bytes | None

Get a blob from storage.

Parameters:

  • key (str) –

    Key associated with the blob to get.

  • default (bytes | None, default: None ) –

    Default return value if the blob does not exist.

Returns:

  • bytes | None –

    The blob associated with the key or the value of default.

Source code in proxystore/endpoint/storage.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def get(
    self,
    key: str,
    default: bytes | None = None,
) -> bytes | None:
    """Get a blob from storage.

    Args:
        key: Key associated with the blob to get.
        default: Default return value if the blob does not exist.

    Returns:
        The blob associated with the key or the value of `default`.
    """
    return self._data.get(key, default)

set() async 

set(key: str, blob: bytes) -> None

Store the blob associated with a key.

Parameters:

  • key (str) –

    Key that will be used to retrieve the blob.

  • blob (bytes) –

    Blob to store.

Raises:

Source code in proxystore/endpoint/storage.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def set(self, key: str, blob: bytes) -> None:
    """Store the blob associated with a key.

    Args:
        key: Key that will be used to retrieve the blob.
        blob: Blob to store.

    Raises:
        ObjectSizeExceededError: If the max object size is configured and
            the data exceeds that size.
    """
    if (
        self._max_object_size is not None
        and len(blob) > self._max_object_size
    ):
        raise ObjectSizeExceededError(
            f'Bytes value has size {bytes_to_readable(len(blob))} which '
            f'exceeds the {bytes_to_readable(self._max_object_size)} '
            'object limit.',
        )
    self._data[key] = blob

close() async 

close() -> None

Clear all stored blobs.

Source code in proxystore/endpoint/storage.py 
146
147
148
async def close(self) -> None:
    """Clear all stored blobs."""
    self._data.clear()

SQLiteStorage 

SQLiteStorage(
    database_path: str | Path = ":memory:",
    *,
    max_object_size: int | None = MAX_OBJECT_SIZE_DEFAULT
)

SQLite storage protocol for blobs.

Parameters:

  • database_path (str | Path, default: ':memory:' ) –

    Path to database file.

  • max_object_size (int | None, default: MAX_OBJECT_SIZE_DEFAULT ) –

    Optional max size in bytes for any single object stored by the endpoint. If exceeded, an error is raised.

Source code in proxystore/endpoint/storage.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def __init__(
    self,
    database_path: str | pathlib.Path = ':memory:',
    *,
    max_object_size: int | None = MAX_OBJECT_SIZE_DEFAULT,
) -> None:
    if database_path == ':memory:':
        self.database_path = database_path
    else:
        path = pathlib.Path(database_path).expanduser().resolve()
        self.database_path = str(path)

    self._max_object_size = max_object_size
    self._db: aiosqlite.Connection | None = None

db() async 

db() -> Connection

Get the database connection object.

Source code in proxystore/endpoint/storage.py 
175
176
177
178
179
180
181
182
183
async def db(self) -> aiosqlite.Connection:
    """Get the database connection object."""
    if self._db is None:
        self._db = await aiosqlite.connect(self.database_path)
        await self._db.execute(
            'CREATE TABLE IF NOT EXISTS blobs'
            '(key TEXT PRIMARY KEY, value BLOB NOT NULL)',
        )
    return self._db

evict() async 

evict(key: str) -> None

Evict a blob from storage.

Parameters:

  • key (str) –

    Key associated with blob to evict.

Source code in proxystore/endpoint/storage.py 
185
186
187
188
189
190
191
192
193
async def evict(self, key: str) -> None:
    """Evict a blob from storage.

    Args:
        key: Key associated with blob to evict.
    """
    db = await self.db()
    await db.execute('DELETE FROM blobs WHERE key=?', (key,))
    await db.commit()

exists() async 

exists(key: str) -> bool

Check if a blob exists in the storage.

Parameters:

  • key (str) –

    Key associated with the blob to check.

Returns:

  • bool –

    If a blob associated with the key exists.

Source code in proxystore/endpoint/storage.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
async def exists(self, key: str) -> bool:
    """Check if a blob exists in the storage.

    Args:
        key: Key associated with the blob to check.

    Returns:
        If a blob associated with the key exists.
    """
    db = await self.db()
    async with db.execute(
        'SELECT count(*) FROM blobs WHERE key=?',
        (key,),
    ) as cursor:
        result = await cursor.fetchone()
        # count() won't ever return 0 rows but mypy doesn't know this
        assert result is not None
        (count,) = result
        return bool(count)

get() async 

get(key: str, default: bytes | None = None) -> bytes | None

Get a blob from storage.

Parameters:

  • key (str) –

    Key associated with the blob to get.

  • default (bytes | None, default: None ) –

    Default return value if the blob does not exist.

Returns:

  • bytes | None –

    The blob associated with the key or the value of default.

Source code in proxystore/endpoint/storage.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
async def get(
    self,
    key: str,
    default: bytes | None = None,
) -> bytes | None:
    """Get a blob from storage.

    Args:
        key: Key associated with the blob to get.
        default: Default return value if the blob does not exist.

    Returns:
        The blob associated with the key or the value of `default`.
    """
    db = await self.db()
    async with db.execute(
        'SELECT value FROM blobs WHERE key=?',
        (key,),
    ) as cursor:
        result = await cursor.fetchone()
        if result is None:
            return default
        else:
            return result[0]

set() async 

set(key: str, blob: bytes) -> None

Store the blob associated with a key.

Parameters:

  • key (str) –

    Key that will be used to retrieve the blob.

  • blob (bytes) –

    Blob to store.

Raises:

Source code in proxystore/endpoint/storage.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
async def set(self, key: str, blob: bytes) -> None:
    """Store the blob associated with a key.

    Args:
        key: Key that will be used to retrieve the blob.
        blob: Blob to store.

    Raises:
        ObjectSizeExceededError: If the max object size is configured and
            the data exceeds that size.
    """
    if (
        self._max_object_size is not None
        and len(blob) > self._max_object_size
    ):
        raise ObjectSizeExceededError(
            f'Bytes value has size {bytes_to_readable(len(blob))} which '
            f'exceeds the {bytes_to_readable(self._max_object_size)} '
            'object limit.',
        )
    db = await self.db()
    await db.execute(
        'INSERT OR REPLACE INTO blobs (key, value) VALUES (?, ?)',
        (key, blob),
    )
    await db.commit()

close() async 

close() -> None

Close the storage.

Source code in proxystore/endpoint/storage.py 
267
268
269
270
async def close(self) -> None:
    """Close the storage."""
    if self._db is not None:
        await self._db.close()