wehub-resource-sync
cddb07a176
docs / deploy (push) Has been cancelled
docs / changes (push) Has been cancelled
docs / check-and-build (push) Has been cancelled
build container image / cpu (push) Has been cancelled
build container image / cuda (push) Has been cancelled
build container image / rocm (push) Has been cancelled
frontend checks / frontend-checks (push) Has been cancelled
frontend tests / frontend-tests (push) Has been cancelled
lfs checks / lfs-check (push) Has been cancelled
python checks / python-checks (push) Has been cancelled
python tests / py3.12: macos-default (push) Has been cancelled
python tests / py3.11: windows-cpu (push) Has been cancelled
python tests / py3.12: windows-cpu (push) Has been cancelled
python tests / py3.11: linux-cpu (push) Has been cancelled
typegen checks / typegen-checks (push) Has been cancelled
uv lock checks / uv-lock-checks (push) Has been cancelled
openapi checks / openapi-checks (push) Has been cancelled
python tests / py3.11: macos-default (push) Has been cancelled
python tests / py3.12: linux-cpu (push) Has been cancelled
73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
from abc import ABC, abstractmethod
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from PIL.Image import Image as PILImageType
|
|
|
|
|
|
class ImageFileStorageBase(ABC):
|
|
"""Low-level service responsible for storing and retrieving image files."""
|
|
|
|
@abstractmethod
|
|
def get(self, image_name: str, image_subfolder: str = "") -> PILImageType:
|
|
"""Retrieves an image as PIL Image."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_path(self, image_name: str, thumbnail: bool = False, image_subfolder: str = "") -> Path:
|
|
"""Gets the internal path to an image or thumbnail."""
|
|
pass
|
|
|
|
@property
|
|
@abstractmethod
|
|
def image_root(self) -> Path:
|
|
"""Gets the root directory for full-size images."""
|
|
pass
|
|
|
|
@property
|
|
@abstractmethod
|
|
def thumbnail_root(self) -> Path:
|
|
"""Gets the root directory for thumbnails."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def evict_cache_paths(self, paths: list[Path]) -> None:
|
|
"""Evicts any cached image objects for the provided paths."""
|
|
pass
|
|
|
|
# TODO: We need to validate paths before starlette makes the FileResponse, else we get a
|
|
# 500 internal server error. I don't like having this method on the service.
|
|
@abstractmethod
|
|
def validate_path(self, path: str) -> bool:
|
|
"""Validates the path given for an image or thumbnail."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def save(
|
|
self,
|
|
image: PILImageType,
|
|
image_name: str,
|
|
metadata: Optional[str] = None,
|
|
workflow: Optional[str] = None,
|
|
graph: Optional[str] = None,
|
|
thumbnail_size: int = 256,
|
|
image_subfolder: str = "",
|
|
) -> None:
|
|
"""Saves an image and a 256x256 WEBP thumbnail. Returns a tuple of the image name, thumbnail name, and created timestamp."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete(self, image_name: str, image_subfolder: str = "") -> None:
|
|
"""Deletes an image and its thumbnail (if one exists)."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_workflow(self, image_name: str, image_subfolder: str = "") -> Optional[str]:
|
|
"""Gets the workflow of an image."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_graph(self, image_name: str, image_subfolder: str = "") -> Optional[str]:
|
|
"""Gets the graph of an image."""
|
|
pass
|