Source code for luna_faces.app.handlers.face_handler

from sanic.response import HTTPResponse

from app.handlers.base_handler import FaceBaseRequestHandler
from app.handlers.helpers import SearchFacesFilters
from app.handlers.query_validators import targetCheckerFactory
from app.handlers.schemas import UPDATE_FACE_SCHEMA
from app.handlers.validators import validate_url
from app.version import VERSION
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
from crutches_on_wheels.maps.vl_maps import FACE_TARGET_MAP
from crutches_on_wheels.web.query_getters import uuidGetter


class FaceHandler(FaceBaseRequestHandler):
    """
    Handler for work with face
    """

    async def head(self, faceId: str) -> HTTPResponse:
        """
        Check the face existence. See `spec_check_face`_.

        .. _spec_check_face:
            _static/api.html#operation/checkFace

        Returns:
            response with status code 200 if the face exists
        """
        accountId = self.getQueryParam("account_id", uuidGetter)
        if not await self.facesContext.isFacesExist([faceId], accountId):
            raise VLException(Error.FaceNotFound.format(faceId), 404, isCriticalError=False)
        return self.success(200, extraHeaders={"Content-Type": "application/json"})

    async def get(self, faceId: str) -> HTTPResponse:
        """
        Get the face. See `spec_get_face`_.

        .. _spec_get_face:
            _static/api.html#operation/getFace

        Returns:
            response json with the face
        """
        accountId = self.getQueryParam("account_id", uuidGetter)
        filters = SearchFacesFilters(account_id=accountId, face_ids=[faceId])
        targets = self.getQueryParam(
            "targets", targetCheckerFactory(list(FACE_TARGET_MAP)), default=list(FACE_TARGET_MAP)
        )
        faces = await self.facesContext.getFaces(filters=filters, targets=targets)
        if len(faces):
            return self.success(200, outputJson=faces[0])
        else:
            return self.error(404, error=Error.FaceNotFound.format(faceId))

    async def delete(self, faceId: str) -> HTTPResponse:
        """
        Delete the face. See `spec_delete_face`_.

        .. _spec_delete_face:
            _static/api.html#operation/deleteFace

        Returns:
            response with status code 204 if the face was successfully deleted
        """
        accountId = self.getQueryParam("account_id", uuidGetter)
        removedFaceCount = await self.facesContext.deleteFaces([faceId], accountId=accountId)
        if not removedFaceCount:
            raise VLException(Error.FaceNotFound.format(faceId), 404, isCriticalError=False)
        return self.success(204)

    async def patch(self, faceId: str) -> HTTPResponse:
        """
        Update the face. See `spec_patch_face`_.

        .. _spec_patch_face:
            _static/api.html#operation/patchFace

        Returns:
            response with status code 204 if the face was successfully updated
        """
        accountId = self.getQueryParam("account_id", uuidGetter)
        data = self.request.json
        self.validateJson(data, UPDATE_FACE_SCHEMA, False)
        validate_url(data.get("avatar"))
        updatedFaceCount = await self.facesContext.updateFace(faceId=faceId, accountId=accountId, **data)
        if not updatedFaceCount:
            raise VLException(Error.FaceNotFound.format(faceId), 404, isCriticalError=False)
        return self.success(204)

    async def put(self, faceId: str) -> HTTPResponse:
        """
        Create the face. See `spec_put_face`_.

        .. _spec_put_face:
            _static/api.html#operation/putFace

        Returns:
            response json with new face id
        """
        newFace = await self.getDataForNewFace()
        await self.facesContext.createFace(faceId, listIds=newFace.listIds, attribute=newFace.attribute, **newFace.data)
        location = f"/{VERSION['Version']['api']}/faces/{faceId}"
        self.respHeaders["Location"] = location
        return self.success(200, outputJson={"face_id": faceId, "url": location})