Skip to main content

ObjectStoragePort

What it is​

Defines abstract interfaces (ports) for an object storage service:

  • Adapter port for infrastructure implementations (e.g., S3, filesystem).
  • Domain port for application/domain layer usage. Also defines storage-related exception types.

Public API​

Exceptions​

Container for exception types:

  • Exceptions.ObjectNotFound: Raised when an object cannot be found.
  • Exceptions.ObjectAlreadyExists: Raised when attempting to create an object that already exists.

IObjectStorageAdapter (abstract base class)​

Interface to be implemented by object storage adapters.

Methods:

  • get_object(prefix: str, key: str) -> bytes: Retrieve object content.
  • put_object(prefix: str, key: str, content: bytes) -> None: Store object content.
  • delete_object(prefix: str, key: str) -> None: Delete an object.
  • list_objects(prefix: str, queue: Optional[Queue] = None) -> list[str]: List object keys under a prefix; may optionally use a Queue.

IObjectStorageDomain (abstract base class)​

Interface for the domain/application layer; mirrors the adapter API.

Methods:

  • get_object(prefix: str, key: str) -> bytes
  • put_object(prefix: str, key: str, content: bytes) -> None
  • delete_object(prefix: str, key: str) -> None
  • list_objects(prefix: str, queue: Optional[Queue] = None) -> list[str]

Configuration/Dependencies​

  • Python standard library:
    • abc.ABC, abc.abstractmethod
    • queue.Queue
    • typing.Optional

No runtime configuration is defined in this module.

Usage​

Implement one of the interfaces and use it through the abstract type.

from queue import Queue
from naas_abi_core.services.object_storage.ObjectStoragePort import (
    IObjectStorageAdapter, Exceptions
)

class InMemoryObjectStorage(IObjectStorageAdapter):
    def __init__(self):
        self._store = {}

    def get_object(self, prefix: str, key: str) -> bytes:
        k = (prefix, key)
        if k not in self._store:
            raise Exceptions.ObjectNotFound()
        return self._store[k]

    def put_object(self, prefix: str, key: str, content: bytes) -> None:
        self._store[(prefix, key)] = content

    def delete_object(self, prefix: str, key: str) -> None:
        self._store.pop((prefix, key), None)

    def list_objects(self, prefix: str, queue: Queue | None = None) -> list[str]:
        keys = [k for (p, k) in self._store.keys() if p == prefix]
        if queue is not None:
            for k in keys:
                queue.put(k)
        return keys

storage = InMemoryObjectStorage()
storage.put_object("docs", "readme.txt", b"hello")
print(storage.get_object("docs", "readme.txt"))
print(storage.list_objects("docs"))

Caveats​

  • These are interfaces only; no concrete storage behavior is provided.
  • list_objects(..., queue=...) accepts an optional Queue, but how/if it is used is implementation-defined.