Source code for luna_backport3.app.handlers.liveness_handler

from luna3.common.luna_response import LunaResponse
from sanic.response import HTTPResponse

from app.handlers.base_handler import APIProxyBaseHandler
from crutches_on_wheels.errors.errors import ErrorInfo
from crutches_on_wheels.errors.exception import VLException
from crutches_on_wheels.web.base_proxy_handler_class import ProxyRequest


class LivenessHandler(APIProxyBaseHandler):
    """
    Handler for liveness prediction service
    """

    allowedMethods = ("POST",)

    async def prepareRequestPost(self) -> ProxyRequest:
        """
        Predict liveness, see `spec liveness`_.

        .. _`spec liveness`:
            _static/api.html#tag/liveness

        Returns:
            Proxy request with new face
        """
        headers = self.prepareHeaders()
        headers.update({"Luna-Account-Id": self.accountId})
        return ProxyRequest(self.request.body, headers, self.prepareQuery())

    async def postProcessingPost(self, response: LunaResponse) -> HTTPResponse:
        """
        Default post processing response from the service
        Args:
            response: response

        Returns:
            response in api format
        """
        responseJson = response.json
        for imageDict in responseJson["images"]:
            if (error := imageDict.get("error")) is not None:
                imageDict["error"] = {"error_code": error["error_code"], "detail": error["detail"]}
        return self.success(response.statusCode, outputJson=responseJson, extraHeaders=response.headers)

    async def postProcessingFailedRequest(self, response: LunaResponse) -> HTTPResponse:
        """
        Post processing failed response from the service
        Args:
            response: response

        Returns:
            response in api format
        """
        raise VLException(ErrorInfo.fromDict(response.json), response.statusCode, False)