83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
"""Definition of ImageRepository class."""
|
|
|
|
from __future__ import annotations
|
|
from pathlib import Path
|
|
from io import BytesIO
|
|
from uuid import UUID
|
|
from PIL import Image
|
|
|
|
from python_repositories.adapters import MinioAdapter
|
|
|
|
from data_store.interfaces import ImageRepositoryInterface
|
|
|
|
|
|
class ImageRepository(MinioAdapter, ImageRepositoryInterface):
|
|
"""ImageRepository implementation using MinIO as the backend."""
|
|
|
|
image_format: str = "PNG"
|
|
folder_name: str = "images"
|
|
|
|
@classmethod
|
|
def _object_name(cls, image_id: UUID) -> str:
|
|
"""Generate the object name for a given image ID."""
|
|
# Check input
|
|
if not isinstance(image_id, UUID):
|
|
raise ValueError("image_id must be a valid UUID.")
|
|
# Return object name
|
|
return f"{cls.folder_name}/{image_id}.{cls.image_format.lower()}"
|
|
|
|
def __enter__(self) -> ImageRepository:
|
|
"""Enter the runtime context related to this object."""
|
|
super().__enter__()
|
|
return self
|
|
|
|
def get(self, image_id: UUID) -> Image.Image | None:
|
|
"""Retrieve an Image by its ID."""
|
|
# Check input
|
|
if not isinstance(image_id, UUID):
|
|
raise ValueError("image_id must be a valid UUID.")
|
|
# Get object
|
|
object_name = self._object_name(image_id)
|
|
buffer = self._get(object_name)
|
|
if buffer is None:
|
|
return None
|
|
# Convert bytes back to PIL Image
|
|
image: Image.Image = Image.open(buffer, formats=[self.image_format])
|
|
return image
|
|
|
|
def put(self, image: Image.Image, image_id: UUID) -> None:
|
|
"""Update or create an Image."""
|
|
# Check inputs
|
|
if not isinstance(image, Image.Image):
|
|
raise ValueError("image must be a PIL Image instance.")
|
|
if not isinstance(image_id, UUID):
|
|
raise ValueError("image_id must be a valid UUID.")
|
|
# Convert image to bytes
|
|
buffer = BytesIO()
|
|
image.save(buffer, self.image_format)
|
|
# Put object
|
|
object_name = self._object_name(image_id)
|
|
self._put(object_name, buffer)
|
|
|
|
def delete(self, image_id: UUID) -> None:
|
|
"""Delete an Image by its ID."""
|
|
# Check input
|
|
if not isinstance(image_id, UUID):
|
|
raise ValueError("image_id must be a valid UUID.")
|
|
# Delete object
|
|
object_name = self._object_name(image_id)
|
|
self._delete(object_name)
|
|
|
|
def list_all(self) -> list[UUID]:
|
|
"""List all image IDs in the repository."""
|
|
# List objects in the bucket
|
|
objects = self._list_objects(prefix=self.folder_name + "/")
|
|
# Stop early if no objects
|
|
if len(objects) == 0:
|
|
return []
|
|
# Remove folder prefix and file extension
|
|
object_stems = [Path(obj).stem for obj in objects]
|
|
# Convert to list of UUIDs
|
|
image_ids = [UUID(name) for name in object_stems]
|
|
return image_ids
|