143 lines
4.0 KiB
Python
143 lines
4.0 KiB
Python
"""Image repository for database operations."""
|
|
|
|
from collections.abc import Sequence
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database.models.board_image import BoardImage
|
|
from app.database.models.image import Image
|
|
|
|
|
|
class ImageRepository:
|
|
"""Repository for image database operations."""
|
|
|
|
def __init__(self, db: Session):
|
|
"""Initialize repository with database session."""
|
|
self.db = db
|
|
|
|
def create_image(
|
|
self,
|
|
user_id: UUID,
|
|
filename: str,
|
|
storage_path: str,
|
|
file_size: int,
|
|
mime_type: str,
|
|
width: int,
|
|
height: int,
|
|
metadata: dict,
|
|
) -> Image:
|
|
"""Create new image record."""
|
|
image = Image(
|
|
user_id=user_id,
|
|
filename=filename,
|
|
storage_path=storage_path,
|
|
file_size=file_size,
|
|
mime_type=mime_type,
|
|
width=width,
|
|
height=height,
|
|
image_metadata=metadata,
|
|
)
|
|
self.db.add(image)
|
|
self.db.commit()
|
|
self.db.refresh(image)
|
|
return image
|
|
|
|
def get_image_by_id(self, image_id: UUID) -> Image | None:
|
|
"""Get image by ID."""
|
|
return self.db.query(Image).filter(Image.id == image_id).first()
|
|
|
|
def get_user_images(
|
|
self, user_id: UUID, limit: int = 50, offset: int = 0
|
|
) -> tuple[Sequence[Image], int]:
|
|
"""Get all images for a user with pagination."""
|
|
total = self.db.query(Image).filter(Image.user_id == user_id).count()
|
|
images = (
|
|
self.db.query(Image)
|
|
.filter(Image.user_id == user_id)
|
|
.order_by(Image.created_at.desc())
|
|
.limit(limit)
|
|
.offset(offset)
|
|
.all()
|
|
)
|
|
return images, total
|
|
|
|
def delete_image(self, image_id: UUID) -> bool:
|
|
"""Delete image record."""
|
|
image = self.get_image_by_id(image_id)
|
|
if not image:
|
|
return False
|
|
|
|
self.db.delete(image)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def increment_reference_count(self, image_id: UUID) -> None:
|
|
"""Increment reference count for image."""
|
|
image = self.get_image_by_id(image_id)
|
|
if image:
|
|
image.reference_count += 1
|
|
self.db.commit()
|
|
|
|
def decrement_reference_count(self, image_id: UUID) -> int:
|
|
"""Decrement reference count for image."""
|
|
image = self.get_image_by_id(image_id)
|
|
if image and image.reference_count > 0:
|
|
image.reference_count -= 1
|
|
self.db.commit()
|
|
return image.reference_count
|
|
return 0
|
|
|
|
def add_image_to_board(
|
|
self,
|
|
board_id: UUID,
|
|
image_id: UUID,
|
|
position: dict,
|
|
transformations: dict,
|
|
z_order: int = 0,
|
|
) -> BoardImage:
|
|
"""Add image to board."""
|
|
board_image = BoardImage(
|
|
board_id=board_id,
|
|
image_id=image_id,
|
|
position=position,
|
|
transformations=transformations,
|
|
z_order=z_order,
|
|
)
|
|
self.db.add(board_image)
|
|
|
|
# Increment reference count
|
|
self.increment_reference_count(image_id)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(board_image)
|
|
return board_image
|
|
|
|
def get_board_images(self, board_id: UUID) -> Sequence[BoardImage]:
|
|
"""Get all images for a board, ordered by z-order."""
|
|
return (
|
|
self.db.query(BoardImage)
|
|
.filter(BoardImage.board_id == board_id)
|
|
.order_by(BoardImage.z_order.asc())
|
|
.all()
|
|
)
|
|
|
|
def remove_image_from_board(self, board_id: UUID, image_id: UUID) -> bool:
|
|
"""Remove image from board."""
|
|
board_image = (
|
|
self.db.query(BoardImage)
|
|
.filter(BoardImage.board_id == board_id, BoardImage.image_id == image_id)
|
|
.first()
|
|
)
|
|
|
|
if not board_image:
|
|
return False
|
|
|
|
self.db.delete(board_image)
|
|
|
|
# Decrement reference count
|
|
self.decrement_reference_count(image_id)
|
|
|
|
self.db.commit()
|
|
return True
|