Brian Bjarke JensenandCursor
5311d49fa6
PR Title Check / check-title (pull_request) Successful in 6s
Test Python Package / integration-tests (pull_request) Successful in 1m17s
Test Python Package / unit-tests (pull_request) Successful in 1m28s
Code Quality Pipeline / code-quality (pull_request) Successful in 1m36s
Test Python Package / coverage-report (pull_request) Successful in 16s
Return None only for missing objects and re-raise other S3 and network failures so callers can distinguish not-found from real errors. Co-authored-by: Cursor <[email protected]>
38 lines
1000 B
Python
38 lines
1000 B
Python
"""Definition of ObjectRepositoryInterface abstract base class."""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from io import BytesIO
|
|
|
|
|
|
class ObjectRepositoryInterface(ABC):
|
|
"""Interface that defines binary object CRUD methods."""
|
|
|
|
@abstractmethod
|
|
def get(self, object_name: str) -> BytesIO | None:
|
|
"""Get an object by name.
|
|
|
|
Returns None when the object does not exist. Raises ConnectionError when
|
|
not connected. Other backend errors propagate to the caller.
|
|
"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def put(
|
|
self,
|
|
object_name: str,
|
|
data: BytesIO,
|
|
content_type: str = "application/octet-stream",
|
|
) -> None:
|
|
"""Put an object by name."""
|
|
...
|
|
|
|
@abstractmethod
|
|
def delete(self, object_name: str) -> None:
|
|
"""Delete an object by name."""
|
|
...
|
|
|
|
@abstractmethod
|
|
def list_objects(self, prefix: str = "") -> list[str]:
|
|
"""List object names with an optional prefix."""
|
|
...
|