79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Definition of JobRepository class."""
|
|
|
|
from __future__ import annotations
|
|
from uuid import UUID
|
|
from python_repositories.adapters import MinioAdapter
|
|
from data_store.interfaces import JobRepositoryInterface
|
|
from data_store.repositories import AnnotationRepository, ImageRepository
|
|
from data_store.dto import Job
|
|
|
|
|
|
class JobRepository(MinioAdapter, JobRepositoryInterface):
|
|
"""JobRepository implementation using MinIO as the backend."""
|
|
|
|
def __enter__(self) -> JobRepository:
|
|
"""Enter the runtime context related to this object."""
|
|
super().__enter__()
|
|
return self
|
|
|
|
def get(self, job_id: UUID) -> Job | None:
|
|
"""Retrieve a Job by its ID."""
|
|
# Check input
|
|
if not isinstance(job_id, UUID):
|
|
raise ValueError("job_id must be a valid UUID.")
|
|
# Get image
|
|
with ImageRepository() as image_repo:
|
|
image = image_repo.get(job_id)
|
|
if image is None:
|
|
self.logger.warning(f"Image for job ID {job_id} not found.")
|
|
return None
|
|
# Get annotation
|
|
with AnnotationRepository() as annotation_repo:
|
|
annotation = annotation_repo.get(job_id)
|
|
if annotation is None:
|
|
self.logger.warning(f"Annotation for job ID {job_id} not found.")
|
|
return None
|
|
# Create Job DTO
|
|
job = Job(
|
|
job_id=job_id,
|
|
image=image,
|
|
annotation=annotation,
|
|
)
|
|
return job
|
|
|
|
def put(self, job: Job) -> None:
|
|
"""Update or create a Job."""
|
|
# Check input
|
|
if not isinstance(job, Job):
|
|
raise ValueError("job must be a Job instance.")
|
|
# Store image
|
|
with ImageRepository() as image_repo:
|
|
image_repo.put(job.image, job.job_id)
|
|
# Store annotation
|
|
with AnnotationRepository() as annotation_repo:
|
|
annotation_repo.put(job.annotation, job.job_id)
|
|
|
|
def delete(self, job_id: UUID) -> None:
|
|
"""Delete a Job by its ID."""
|
|
# Check input
|
|
if not isinstance(job_id, UUID):
|
|
raise ValueError("job_id must be a valid UUID.")
|
|
# Delete image
|
|
with ImageRepository() as image_repo:
|
|
image_repo.delete(job_id)
|
|
# Delete annotation
|
|
with AnnotationRepository() as annotation_repo:
|
|
annotation_repo.delete(job_id)
|
|
|
|
def list_all(self) -> list[UUID]:
|
|
"""List all Job IDs."""
|
|
# List image IDs
|
|
with ImageRepository() as image_repo:
|
|
image_ids = set(image_repo.list_all())
|
|
# List annotation IDs
|
|
with AnnotationRepository() as annotation_repo:
|
|
annotation_ids = set(annotation_repo.list_all())
|
|
# Find intersection of IDs
|
|
job_ids = list(image_ids.intersection(annotation_ids))
|
|
return job_ids
|