File size: 6,517 Bytes
985c397 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | # SPDX-License-Identifier: LGPL-2.1-or-later
# ***************************************************************************
# * Copyright (c) 2025 Samuel Abels <knipknap@gmail.com> *
# * *
# * This program is free software; you can redistribute it and/or modify *
# * it under the terms of the GNU Lesser General Public License (LGPL) *
# * as published by the Free Software Foundation; either version 2 of *
# * the License, or (at your option) any later version. *
# * for detail see the LICENCE text file. *
# * *
# * This program is distributed in the hope that it will be useful, *
# * but WITHOUT ANY WARRANTY; without even the implied warranty of *
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
# * GNU Library General Public License for more details. *
# * *
# * You should have received a copy of the GNU Library General Public *
# * License along with this program; if not, write to the Free Software *
# * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 *
# * USA *
# * *
# ***************************************************************************
import abc
from typing import List
from ..uri import AssetUri
class AssetStore(abc.ABC):
"""
Abstract base class for storing and retrieving asset data as raw bytes.
Stores are responsible for handling the low-level interaction with a
specific storage backend (e.g., local filesystem, HTTP server) based
on the URI protocol.
"""
def __init__(self, name: str, *args, **kwargs):
self.name = name
@abc.abstractmethod
async def get(self, uri: AssetUri) -> bytes:
"""
Retrieve the raw byte data for the asset at the given URI.
Args:
uri: The unique identifier for the asset.
Returns:
The raw byte data of the asset.
Raises:
FileNotFoundError: If the asset does not exist at the URI.
# Other store-specific exceptions may be raised.
"""
raise NotImplementedError
async def exists(self, uri: AssetUri) -> bool:
"""
Check if the asset exists at the given URI.
Args:
uri: The unique identifier for the asset.
Returns:
True if the asset exists, False otherwise.
"""
try:
await self.get(uri)
return True
except FileNotFoundError:
return False
@abc.abstractmethod
async def delete(self, uri: AssetUri) -> None:
"""
Delete the asset at the given URI.
Args:
uri: The unique identifier for the asset to delete.
Raises:
FileNotFoundError: If the asset does not exist at the URI.
# Other store-specific exceptions may be raised.
"""
raise NotImplementedError
@abc.abstractmethod
async def create(self, asset_type: str, asset_id: str, data: bytes) -> AssetUri:
"""
Create a new asset in the store with the given data.
The store determines the final URI for the new asset. The
`asset_type` can be used to influence the storage location
or URI structure (e.g., as part of the path).
Args:
asset_type: The type of the asset (e.g., 'material',
'toolbitshape').
asset_id: The unique identifier for the asset.
data: The raw byte data of the asset to create.
Returns:
The URI of the newly created asset.
Raises:
# Store-specific exceptions may be raised (e.g., write errors).
"""
raise NotImplementedError
@abc.abstractmethod
async def update(self, uri: AssetUri, data: bytes) -> AssetUri:
"""
Update the asset at the given URI with new data, creating a new version.
Args:
uri: The unique identifier of the asset to update.
data: The new raw byte data for the asset.
Raises:
FileNotFoundError: If the asset does not exist at the URI.
# Other store-specific exceptions may be raised (e.g., write errors).
"""
raise NotImplementedError
@abc.abstractmethod
async def list_assets(
self, asset_type: str | None = None, limit: int | None = None, offset: int | None = None
) -> List[AssetUri]:
"""
List assets in the store, optionally filtered by asset type and
with pagination. For versioned stores, this lists the latest
version of each asset.
Args:
asset_type: Optional filter for asset type.
limit: Maximum number of assets to return.
offset: Number of assets to skip from the beginning.
Returns:
A list of URIs for the assets.
"""
raise NotImplementedError
@abc.abstractmethod
async def count_assets(self, asset_type: str | None = None) -> int:
"""
Counts assets in the store, optionally filtered by asset type.
Args:
asset_type: Optional filter for asset type.
Returns:
The number of assets.
"""
raise NotImplementedError
@abc.abstractmethod
async def list_versions(self, uri: AssetUri) -> List[AssetUri]:
"""
Lists available version identifiers for a specific asset URI.
Args:
uri: The URI of the asset (version component is ignored).
Returns:
A list of URIs pointing to the specific versions of the asset.
"""
raise NotImplementedError
@abc.abstractmethod
async def is_empty(self, asset_type: str | None = None) -> bool:
"""
Checks if the store contains any assets, optionally filtered by asset
type.
Args:
asset_type: Optional filter for asset type.
Returns:
True if the store is empty (or empty for the given asset type),
False otherwise.
"""
raise NotImplementedError
|