Source code for luna_faces.app.handlers.faces_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 DELETE_FACES_SCHEMA
from app.version import VERSION
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.maps.vl_maps import FACE_TARGET_MAP
from crutches_on_wheels.web.query_getters import int01Getter, uuidGetter


class FacesHandler(FaceBaseRequestHandler):
    """
    Faces handler.
    """

    async def post(self) -> HTTPResponse:
        """
        Create a new face. See `spec_create_face`_.

        .. _spec_create_face:
            _static/api.html#operation/createFace

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

    async def get(self) -> HTTPResponse:
        """
        Get faces by filters. See `spec_get_faces`_.

        .. _spec_get_faces:
            _static/api.html#operation/getFaces

        Returns:
            response json with faces
        """
        pageSize = self.getQueryParam("page_size")
        if pageSize == "-1":
            page, pageSize = 1, 10**9
        else:
            page, pageSize = self.getPagination()
        filters = SearchFacesFilters().loadFromQuery(self.getQueryParam)
        targets = self.getQueryParam(
            "targets", targetCheckerFactory(list(FACE_TARGET_MAP)), default=list(FACE_TARGET_MAP)
        )

        faces = await self.facesContext.getFaces(filters=filters, targets=targets, page=page, pageSize=pageSize)
        return self.success(200, outputJson={"faces": faces})

    async def delete(self) -> HTTPResponse:
        """
        Delete several faces. See `spec_delete_faces`_.

        .. _spec_delete_faces:
            _static/api.html#operation/deleteFaces

        Returns:
            response with status code 204 if faces was successfully deleted
            or response with status code 400 if some face was not found
        """
        accountId = self.getQueryParam("account_id", uuidGetter)
        data = self.request.json
        self.validateJson(data, DELETE_FACES_SCHEMA, False)
        faceIds = set(data["face_ids"])
        ignoreFlag = self.getQueryParam("ignore", int01Getter, default=0)

        if not ignoreFlag:
            nonExistFaceId = await self.facesContext.getNonexistentFaceId(faceIds, accountId)
            if nonExistFaceId is not None:
                return self.error(400, error=Error.FacesNotFound.format(nonExistFaceId))

        await self.facesContext.deleteFaces(faceIds)
        return self.success(204)