Source code for services.posts.comment

from typing import TYPE_CHECKING, Any, Dict, List, Optional

from django.conf import settings
from django.db import transaction

from apps.posts.models import Comment
from services.notifications.notification import NotificationService
from services.posts.comment_image import CommentImageService
from services.posts.exceptions import CommentLimitImageException, PostUnavailableForCommentingException

if TYPE_CHECKING:
    from apps.posts.models import CommentImage, Post
    from apps.users.models import User




class CommentService:
    comment_image_service = CommentImageService
    notification_service = NotificationService

    @classmethod
    def _create(cls, data: Dict[str, Any]) -> "Comment":
        return Comment.objects.create(**data)

    @classmethod
    def _validate(cls, data: Dict[str, Any], author: "User", post: Optional["Post"] = None):
        if not post:
            post = data["post"]
        assert post is not None  # noqa
        if (
            post.is_deleted
            or (post.visibility == post.VisibilityChoices.PRIVATE and author != post.author)
            or not post.is_commenting_allowed
        ):
            raise PostUnavailableForCommentingException
        return data



    @classmethod
    @transaction.atomic
    def create(cls, data: Dict[str, Any]) -> "Comment":
        cls._validate(data, data["author"])
        images = data.pop("images", [])
        comment = cls._create(data)
        cls.add_images(comment, images)
        cls.notification_service.send_comment_notification(
            sender=comment.author, receiver=comment.post.author, post=comment.post, data={"comment_id": comment.id}
        )
        return comment




    @classmethod
    def add_images(cls, comment: "Comment", images: List[Dict[str, Any]]) -> None:
        if len(images) > 0:
            for image in images:
                cls.add_image(comment, image)




    @classmethod
    def add_image(cls, comment: "Comment", image) -> "CommentImage":
        cls.validate_count_images(comment)
        data = {"comment_id": comment.pk, "image": image}
        return cls.comment_image_service.create(data)




    @classmethod
    def validate_count_images(cls, comment: "Comment"):
        if comment.images.count() == settings.MAX_COMMENT_ITEMS:
            raise CommentLimitImageException()




    @classmethod
    def update(cls, instance: "Comment", data: Dict[str, Any]) -> "Comment":
        """update for Verification"""
        cls._validate(data, instance.author, instance.post)
        for attr, value in data.items():
            setattr(instance, attr, value)
        instance.save()
        return instance




    @classmethod
    @transaction.atomic
    def delete(cls, instance: "Comment") -> None:
        instance.is_deleted = True
        instance.save(update_fields=["is_deleted"])
        cls.comment_image_service.remove_by_comment(instance)