module emote.extra.crud_storage

Generic CRUD-based storage on disk.

Classes

class StorageItemHandle(Generic[T]):

A handle that represents a storage item. Can be safely exposed to users. Not cryptographically safe: handles are guessable.

You can convert this handle from and to strings using str(handle) and StorageItemHandle.from_string(string).

Fields

  • handle: int

Methods

def from_string(value) -> Optional['StorageItemHandle']

Parses a handle from its string representation. Returns None if the handle is invalid.

Arguments:

  • value(str)

class StorageItem(Generic[T]):

Fields

  • handle: StorageItemHandle[T]

  • timestamp: datetime

  • filepath: str

class CRUDStorage(Generic[T]):

Manages a set of files on disk in a simple CRUD way. All files will be stored to a single directory with a name on the format {prefix}{timestamp}_{index}.{extension}.

This class is thread-safe.

Methods

def __init__(self, directory, prefix, extension) -> None

Arguments:

  • directory(str)
  • prefix(str)
  • extension(str) (default: bin)
def create_with_data(self, data) -> StorageItem[T]

Creates a new file with the given data.

Arguments:

  • data(bytearray)
def create_from_filepath(self, filepath) -> StorageItem[T]

Creates a new entry for an existing file. The file must already be in the directory that this storage manages. It does not need to conform to the naming convention that the CRUDStorage normally uses.

Arguments:

  • filepath(str)
def create_with_saver(self, saver) -> StorageItem[T]

Creates a new file by saving it via the provided function. The function will be called with the path at which the file should be saved.

Arguments:

  • saver(Callable[[str], None])
def update(self, handle, data) -> None

Updates an existing file with the given contents.

Arguments:

  • handle(StorageItemHandle[T])
  • data(bytearray)
def items(self) -> Sequence[StorageItem[T]]

Returns:

  • a sequence of all files owned by this storage.
def delete(self, handle) -> bool

Deletes an existing file owned by this storage.

Arguments:

  • handle(StorageItemHandle[T])

Returns:

  • True if a file was deleted, and false if the file was not owned by this storage.
def get(self, handle) -> Optional[StorageItem[T]]

Arguments:

  • handle(StorageItemHandle[T])

Returns:

  • The storage item corresponding handle or None if it was not found
def latest(self) -> Optional[StorageItem[T]]

The last storage item that was added to the storage. If items have been deleted, this is the last item of the ones that remain.