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 
21
22
23
24
25
26
27
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 
29
30
31
32
33
34
35
36
37
38
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 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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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.

Source code in proxystore/endpoint/storage.py 
56
57
58
59
60
61
62
63
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.
    """
    ...

close() async 

close() -> None

Close the storage.

Source code in proxystore/endpoint/storage.py 
65
66
67
async def close(self) -> None:
    """Close the storage."""
    ...

DictStorage 

DictStorage() -> None

Simple dictionary-based storage for blobs.

Source code in proxystore/endpoint/storage.py 
73
74
def __init__(self) -> None:
    self._data: dict[str, bytes] = {}

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 
76
77
78
79
80
81
82
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 
84
85
86
87
88
89
90
91
92
93
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 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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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.

Source code in proxystore/endpoint/storage.py 
111
112
113
114
115
116
117
118
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.
    """
    self._data[key] = blob

close() async 

close() -> None

Clear all stored blobs.

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

SQLiteStorage 

SQLiteStorage(
    database_path: str | pathlib.Path = ":memory:",
) -> None

SQLite storage protocol for blobs.

Source code in proxystore/endpoint/storage.py 
128
129
130
131
132
133
134
135
def __init__(self, database_path: str | pathlib.Path = ':memory:') -> None:
    if database_path == ':memory:':
        self.database_path = database_path
    else:
        path = pathlib.Path(database_path).expanduser().resolve()
        self.database_path = str(path)

    self._db: aiosqlite.Connection | None = None

db() async 

db() -> aiosqlite.Connection

Get the database connection object.

Source code in proxystore/endpoint/storage.py 
137
138
139
140
141
142
143
144
145
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 
147
148
149
150
151
152
153
154
155
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 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
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:
        (count,) = await cursor.fetchone()
        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 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 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
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.

Source code in proxystore/endpoint/storage.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
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.
    """
    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 
213
214
215
216
async def close(self) -> None:
    """Close the storage."""
    if self._db is not None:
        await self._db.close()