from sanic.response import HTTPResponse
from app.handlers.base_handler import FaceBaseRequestHandler
from app.handlers.helpers import SearchFacesFilters
from app.handlers.query_validators import descriptorVersionGetter, listItemsCheckerFactory
from app.handlers.schemas import FACE_ATTRIBUTE_EXPLICIT_SCHEMA, FACE_ATTRIBUTE_SCHEMA
from crutches_on_wheels.cow.errors.errors import Error
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.maps.vl_maps import TEMPORARY_ATTRIBUTE_TARGET_MAP
from crutches_on_wheels.cow.web.query_getters import uuidGetter
FACE_ATTRIBUTE_TARGET_MAP = [
target for target in TEMPORARY_ATTRIBUTE_TARGET_MAP if target not in ("account_id", "attribute_id")
]
[docs]class FaceAttributesHandler(FaceBaseRequestHandler):
"""
Handler for work with face attributes
"""
[docs] async def getFaceAccount(self, faceId: str) -> str:
"""
Check the face existence and get face account
Args:
faceId: face id
Raises:
VLException(Error.FaceNotFound, 404, isCriticalError=False): if face is not exist
Returns:
face account id
"""
faces = await self.facesContext.getFaces(filters=SearchFacesFilters(face_ids=[faceId]), targets=["account_id"])
if not faces:
raise VLException(Error.FaceNotFound.format(faceId), 404, isCriticalError=False)
return faces[0]["account_id"]
[docs] async def get(self, faceId: str) -> HTTPResponse:
"""
Get the face attribute. See `spec_get_face_attributes`_.
.. _spec_get_face_attributes:
_static/api.html#operation/getFaceAttributes
Returns:
response json with the face attributes
"""
accountId = self.getQueryParam("account_id", uuidGetter)
targets = self.getQueryParam(
"targets",
listItemsCheckerFactory(list(FACE_ATTRIBUTE_TARGET_MAP)),
default=tuple(FACE_ATTRIBUTE_TARGET_MAP),
)
descriptorVersion = self.getQueryParam(
"descriptor_version", descriptorVersionGetter, default=self.config.defaultFaceDescriptorVersion
)
acceptHeaders = self.request.headers.get("Accept", "application/json")
getDescriptorAsBytes = "application/msgpack" in acceptHeaders
faceAttributes = await self.facesContext.getFacesAttributes(
faceIds=[faceId],
retrieveAttrs=targets,
descriptorVersion=descriptorVersion,
accountId=accountId,
getDescriptorAsBytes=getDescriptorAsBytes,
)
if not faceAttributes:
raise VLException(Error.FaceNotFound.format(faceId), 404, isCriticalError=False)
return self.success(200, outputJson={"attributes": faceAttributes[0]["attributes"]})
[docs] async def put(self, faceId: str) -> HTTPResponse:
"""
Put the face attribute. See `spec_put_face_attribute`_.
.. _spec_put_face_attribute:
_static/api.html#operation/putFaceAttribute
Returns:
response with status code 204 if the face attribute was successfully put
"""
accountId = self.getQueryParam("account_id", uuidGetter)
attributeData = self.request.json
self.validateJson(attributeData, FACE_ATTRIBUTE_SCHEMA, False)
# check the license is available
self.checkLicense()
if "attribute_id" in attributeData:
attributeId = attributeData["attribute_id"]
attributeAccountId = accountId or (await self.getFaceAccount(faceId))
attribute = await self.getAttributeById(attributeId=attributeId, accountId=attributeAccountId)
else:
attribute = self.getAttributeByData(attributeData)
# face existence check inside
await self.facesContext.putFaceAttributes(faceId=faceId, attribute=attribute, accountId=accountId)
return self.success(204)
[docs] async def patch(self, faceId: str) -> HTTPResponse:
"""
Patch the face attribute. See `spec_patch_face_attribute`_.
.. _spec_patch_face_attribute:
_static/api.html#operation/patchFaceAttribute
Returns:
response with status code 204 if the face attribute was successfully patched
"""
accountId = self.getQueryParam("account_id", uuidGetter)
attributeData = self.request.json
self.validateJson(attributeData, FACE_ATTRIBUTE_EXPLICIT_SCHEMA, False)
attribute = self.getAttributeByData(attributeData)
# face existence check inside
await self.facesContext.updateFaceAttributes(
faceId=faceId, attribute=attribute, accountId=accountId, forceUpdate=True
)
return self.success(204)
[docs] async def delete(self, faceId: str) -> HTTPResponse:
"""
Delete the face attribute. See `spec_delete_face_attributes`_.
.. _spec_delete_face_attributes:
_static/api.html#operation/deleteFaceAttributes
Returns:
response with status code 204 if the face attribute was successfully deleted
"""
accountId = self.getQueryParam("account_id", uuidGetter)
await self.facesContext.deleteFaceAttributes(faceId, accountId=accountId)
return self.success(204)
[docs]class FaceAttributeSamplesHandler(FaceBaseRequestHandler):
"""
Handler for work with face attribute samples
"""
[docs] async def get(self, faceId: str) -> HTTPResponse:
"""
Get the face attribute samples. See `spec_get_face_attribute_samples`_.
.. _spec_get_face_attribute_samples:
_static/api.html#operation/getFaceAttributeSamples
Returns:
response json with the face attribute samples
"""
accountId = self.getQueryParam("account_id", uuidGetter)
# face existence check inside
sampleIds = await self.facesContext.getFacesAttributeSamples(faceId=faceId, accountId=accountId)
return self.success(200, outputJson={"samples": sampleIds})
