# -*- coding: utf-8 -*-
""" Base handler
Module realize base class for all handlers.
"""
from abc import abstractmethod
from copy import deepcopy
from typing import Any, Dict, NamedTuple, Tuple, Union
import msgpack
import ujson
from luna3.client import Client
from luna3.common.luna_response import LunaResponse
from luna3.common.requests import makeRequest
from multidict import CIMultiDict, CIMultiDictProxy
from sanic.response import HTTPResponse
from stringcase import snakecase
from vlutils.recursive_functions import goDeep
from app.app import ApiApp, ApiRequest
from app.global_vars.context_vars import requestIdCtx
from app.handlers.schemas import schemas
from classes.enums import VisibilityArea
from configs.configs.configs.services import SettingsApi
from configs.configs.configs.settings.classes import (
ImageStoreAddressSettings,
ServiceAddressSettings,
ServiceTimeoutsSettings,
)
from crutches_on_wheels.cow.errors.errors import Error
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.web.base_proxy_handler_class import SessionPool
from crutches_on_wheels.cow.web.handlers import BaseHandler
FACE_SAMPLE_RELATIVE_URL = "/{apiVersion}/samples/faces/{sampleId}"
BODY_SAMPLE_RELATIVE_URL = "/{apiVersion}/samples/bodies/{sampleId}"
ATTRIBUTE_RELATIVE_URL = "/{apiVersion}/attributes/{attributeId}"
EVENT_RELATIVE_URL = "/{apiVersion}/events/{eventId}"
FACE_RELATIVE_URL = "/{apiVersion}/faces/{faceId}"
HEADERS_WHITE_LIST = {
"content-type",
"luna-request-id",
"content-disposition",
"luna-event-time",
"luna-event-end-time",
"accept",
}
[docs]class ProxyRequest(NamedTuple):
"""
Proxy request
"""
body: Any # body for sending to a proxed service
headers: Union[Dict[str, str], CIMultiDictProxy[str]] # headers for sending to a proxed service
query: Dict[str, Union[None, str]] # query for sending to a proxed service
[docs]class BaseRequestHandler(BaseHandler):
"""
Base handler for other handlers.
Attributes:
luna3Client (luna3.client.Client): luna3 client
accountId (str): account id
"""
# IDE type hint stub
request: ApiRequest
def __init__(self, request: ApiRequest):
super().__init__(request)
self.accountId = request.credentials.accountId
self.addDataForMonitoring(tags={"account_id": self.accountId})
requestIdCtx.set(self.requestId)
self.luna3Client: Client = self.app.ctx.luna3session.getClient(self.requestId)
@property
def app(self) -> ApiApp:
"""
Get running app
Returns:
app
"""
return self.request.app
@property
def config(self) -> SettingsApi:
"""
Get app config
Returns:
app config
"""
return self.app.ctx.serviceConfig
[docs] @staticmethod
def filterAllowedHeaders(headers: CIMultiDict[str, str]) -> dict:
"""
Filter only allowed headers.
Args:
headers: headers to filter
Returns:
filtered headers
"""
return {k: v for k, v in headers.items() if k.lower() in HEADERS_WHITE_LIST}
[docs] async def options(self, *args, **kwargs) -> HTTPResponse:
"""
Default 'options' method
"""
allowHeaders = {"Allow": ", ".join(self.getAllowedMethods())}
return self.success(body=None, extraHeaders=allowHeaders)
[docs]class BaseProxyHandler(BaseRequestHandler):
"""
Base proxy handler for other handlers.
"""
# list of allowed proxy methods
allowedMethods: Tuple[str, ...] = ()
# session pool
session = SessionPool()
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a proxed service address
Returns:
a proxed service address
"""
raise NotImplemented
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a proxed service timeouts
Returns:
a proxed service timeouts
"""
raise NotImplemented
@property
def serviceUrl(self) -> str:
"""Service url"""
return f"{self.serviceAddress.origin}/{self.serviceAddress.apiVersion}"
[docs] def prepareRequestDefault(self) -> ProxyRequest:
"""
Prepare proxy request by default
Returns:
default proxy request
"""
return ProxyRequest(self.request.body, self.prepareHeaders(), self.prepareQuery())
[docs] async def prepareRequestPost(self):
"""
Prepare proxy request for method `post`
Returns:
proxy request
"""
return self.prepareRequestDefault()
[docs] async def prepareRequestPatch(self):
"""
Prepare proxy request for method `patch`
Returns:
proxy request
"""
return self.prepareRequestDefault()
[docs] async def prepareRequestDelete(self):
"""
Prepare proxy request for method `delete`
Returns:
proxy request
"""
return self.prepareRequestDefault()
[docs] async def prepareRequestPut(self):
"""
Prepare proxy request for method `put`
Returns:
proxy request
"""
return self.prepareRequestDefault()
[docs] async def prepareProxyRequest(self) -> ProxyRequest:
"""
Prepare proxy request
Returns:
proxy request
"""
if self.request.method in ("GET", "OPTIONS", "HEAD"):
return ProxyRequest(b"", self.prepareHeaders(), self.prepareQuery())
elif self.request.method in ("POST", "PUT", "DELETE", "PATCH"):
return await getattr(self, f"prepareRequest{self.request.method.capitalize()}")()
raise RuntimeError(f"bad request method {self.request.method}")
[docs] def postProcessingDefault(self, response: LunaResponse) -> HTTPResponse:
"""
Default post processing response from the service
Args:
response: response
Returns:
response in api format
"""
headers = self.filterAllowedHeaders(response.headers)
return self.success(response.statusCode, body=response.body, extraHeaders=headers)
[docs] async def postProcessingPost(self, response: LunaResponse) -> HTTPResponse:
"""
Default post processing response from the service
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessingGet(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response from the service for method `get`
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessingPatch(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response from the service for method `patch`
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessingDelete(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response from the service for method `delete`
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessingPut(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response from the service for method `put`
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessingOptions(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response from the service for method `post`
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessingHead(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response from the service for method `head`
Args:
response: response
Returns:
response in api format
"""
return self.postProcessingDefault(response)
[docs] async def postProcessing(self, response: LunaResponse) -> HTTPResponse:
"""
Post processing response
Args:
response: response
Returns:
response in api format
"""
return await getattr(self, f"postProcessing{self.request.method.capitalize()}")(response)
[docs] def prepareRequestCreation(self) -> ProxyRequest:
"""
Add account id to root json for creating some objects
Returns:
proxy request
Raises:
VLException(Error.BadContentType): if request has incorrect content type
"""
headers = self.prepareHeaders()
contentType = self.request.content_type
if contentType == "application/json":
inputJson = self.request.json
elif contentType == "application/octet-stream":
inputJson = {}
headers["Content-Type"] = "application/json"
elif contentType == "application/msgpack":
inputJson = self.request.json
headers["Content-Type"] = "application/msgpack"
else:
raise VLException(Error.BadContentType, statusCode=400, isCriticalError=False)
self.validateJson(inputJson, schemas.OBJECT_SCHEMA)
if contentType == "application/msgpack":
body = msgpack.packb(inputJson, use_bin_type=True)
else:
body = ujson.dumps(inputJson, ensure_ascii=False)
return ProxyRequest(body, headers, self.prepareQuery())
[docs] def prepareQuery(self) -> Dict[str, str]:
"""
Prepare headers for a proxy request
"""
return dict(self.request.args)
[docs] def prepareHeaders(self) -> Dict[str, str]:
"""
Prepare headers for a proxy request
Returns:
headers with `Luna-Request-Id` and other input headers
"""
headers = self.filterAllowedHeaders(self.request.headers)
headers["Luna-Request-Id"] = self.requestId
headers["Accept-Encoding"] = "identity, deflate, gzip"
if (accountId := self.request.credentials.accountId) is not None:
headers["Luna-Account-Id"] = accountId
return headers
[docs] def convertIncomingUrls(self, response: LunaResponse) -> Dict[str, str]:
"""
Convert service URLs to output JSON and update Location in handler
Returns:
output JSON with converted service URLs
"""
outputJson = deepcopy(response.json)
location = outputJson["url"].replace(f"/{self.serviceAddress.apiVersion}", f"/{self.app.ctx.apiVersion}", 1)
self.respHeaders["Location"] = location
outputJson["url"] = location
return outputJson
[docs] def prepareUrl(self) -> str:
"""
Prepare url to a service
Returns:
same url with correct api version
"""
uriWithoutApiVersion = self.request.server_path.replace(f"/{self.app.ctx.apiVersion}", "", 1)
return f"{self.serviceUrl}{uriWithoutApiVersion}"
[docs] def checkTokenPermissionsDefault(self, objectName: str) -> None:
"""
Check token permissions for request using default method to permission map
"""
permissions = self.request.credentials.permissions
if permissions is None:
return
objectPermissions = permissions.__getattribute__(objectName)
action = {
"POST": "creation",
"GET": "view",
"HEAD": "view",
"PUT": "modification",
"PATCH": "modification",
"DELETE": "deletion",
}.get(self.request.method)
if action is not None and action not in objectPermissions:
raise VLException(Error.ForbiddenByToken.format(action, snakecase(objectName)), 403, False)
[docs] @abstractmethod
def checkTokenPermissions(self) -> None:
"""
Check token permissions for request
"""
[docs] async def makeRequestToService(self) -> HTTPResponse:
"""
Make request to custom service
Returns:
response
"""
if self.request.method not in self.allowedMethods:
return self.setMethodNotAllowed()
self.checkTokenPermissions()
url = self.prepareUrl()
preparedRequest = await self.prepareProxyRequest()
kwargs = dict(
url=url,
method=self.request.method,
queryParams=preparedRequest.query,
headers=preparedRequest.headers,
asyncRequest=True,
totalTimeout=self.serviceTimeouts.totalTimeout,
connectTimeout=self.serviceTimeouts.connectTimeout,
sockConnectTimeout=self.serviceTimeouts.sockConnectTimeout,
sockReadTimeout=self.serviceTimeouts.sockReadTimeout,
session=self.session,
)
reply = await makeRequest(body=preparedRequest.body, **kwargs)
if reply.success:
return await self.postProcessing(reply)
return self.postProcessingDefault(reply)
[docs] async def post(self, *args, **kwargs) -> HTTPResponse:
"""
Default 'post' method
Returns:
proxy response
"""
return await self.makeRequestToService()
[docs] async def put(self, *args, **kwargs) -> HTTPResponse:
"""
Default 'put' method
Returns:
proxy response
"""
return await self.makeRequestToService()
[docs] async def get(self, *args, **kwargs) -> HTTPResponse:
"""
Default 'get' method
Returns:
proxy response
"""
return await self.makeRequestToService()
[docs] async def delete(self, *args, **kwargs) -> HTTPResponse:
"""
Default 'delete' method
Returns:
proxy response
"""
return await self.makeRequestToService()
[docs] async def patch(self, *args, **kwargs) -> HTTPResponse:
"""
Default 'patch' method
Returns:
proxy response
"""
return await self.makeRequestToService()
[docs] async def head(self, *args, **kwargs):
"""
Default 'head' method
Returns:
proxy response
"""
return await self.makeRequestToService()
[docs]class EventServiceBaseHandler(BaseProxyHandler):
"""
Class for proxy handlers to events
"""
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a events service address
Returns:
a events service address
"""
return self.config.eventsAddress
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a events service timeouts
Returns:
a events service timeouts
"""
return self.config.eventsTimeouts
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if not self.config.additionalServicesUsage.lunaEvents:
raise VLException(Error.LunaEventsIsDisabled, statusCode=403, isCriticalError=False)
[docs]class FacesServiceBaseHandler(BaseProxyHandler):
"""
Class for proxy handlers to faces
"""
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a faces service address
Returns:
a faces service address
"""
return self.config.facesAddress
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a faces service timeouts
Returns:
a faces service timeouts
"""
return self.config.facesTimeouts
[docs]class HandlersServiceBaseHandler(BaseProxyHandler):
"""
Class for proxy handlers to handlers
"""
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a handlers service address
Returns:
a handlers service address
"""
return self.config.handlersAddress
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a handlers service timeouts
Returns:
a handlers service timeouts
"""
return self.config.handlersTimeouts
[docs] def convertAttributeUrls(self, replyJson: dict) -> dict:
"""
Convert attribute URLs in output JSON
Args:
replyJson: JSON
Returns:
output JSON with converted attribute URLs
"""
outputJson = deepcopy(replyJson)
for attribute in outputJson:
if attribute["attribute_id"]:
attribute["url"] = ATTRIBUTE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
attributeId=attribute["attribute_id"],
)
return outputJson
[docs] def convertVerifierUrls(self, replyJson: dict) -> dict:
"""
Convert attribute and sample URLs in output JSON
Args:
replyJson: JSON
Returns:
output JSON with converted attribute URLs
"""
outputJson = deepcopy(replyJson)
for image in outputJson["images"]:
for detection in image["detections"]["face_detections"]:
attributeId = detection["face_attributes"]["attribute_id"]
if attributeId is not None:
detection["face_attributes"]["url"] = ATTRIBUTE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
attributeId=attributeId,
)
if (sample := detection.get("sample")) is not None:
if sample["face"]["sample_id"] is not None:
detection["sample"]["face"]["url"] = FACE_SAMPLE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
sampleId=sample["face"]["sample_id"],
)
return outputJson
[docs] def convertDetectorSampleUrls(self, replyJson: dict) -> dict:
"""
Convert detections sample URLs in output JSON
Args:
replyJson: JSON
Returns:
output JSON with converted sample URLs
"""
outputJson = deepcopy(replyJson)
for image in outputJson["images"]:
for sample in image["detections"]["samples"]:
sample["face"]["url"] = FACE_SAMPLE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
sampleId=sample["face"]["sample_id"],
)
return outputJson
[docs] def convertHandlerEventsUrls(self, replyJson: dict) -> dict:
"""
Convert events/faces/detections samples URLs in output JSON
Args:
replyJson: JSON
Returns:
output JSON with converted sample URLs
"""
outputJson = deepcopy(replyJson)
for event in outputJson["events"]:
if event["url"] is not None:
event["url"] = EVENT_RELATIVE_URL.format(apiVersion=self.app.ctx.apiVersion, eventId=event["event_id"])
if event["face"] is not None:
event["face"]["url"] = FACE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion, faceId=event["face"]["face_id"]
)
if event["face_attributes"] is not None and event["face_attributes"]["url"] is not None:
event["face_attributes"]["url"] = ATTRIBUTE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
attributeId=event["face_attributes"]["attribute_id"],
)
for detection in event["detections"]:
samples = detection["samples"]
if samples["face"] and samples["face"]["url"]:
samples["face"]["url"] = FACE_SAMPLE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
sampleId=samples["face"]["sample_id"],
)
if samples["body"] and samples["body"]["url"]:
samples["body"]["url"] = BODY_SAMPLE_RELATIVE_URL.format(
apiVersion=self.app.ctx.apiVersion,
sampleId=samples["body"]["sample_id"],
)
return outputJson
[docs] def updatePolicies(self) -> None:
"""
Update policies:
- if visibility area is limited - update matching candidates account id
"""
if self.request.credentials.visibilityArea == VisibilityArea.account and isinstance(
(matchPolicies := goDeep(self.request.json, ["policies", "match_policy"])), list
):
for matchPolicy in matchPolicies:
if isinstance((candidates := goDeep(matchPolicy, ["candidates"])), dict):
accountId = candidates.get("account_id")
if accountId is None:
candidates["account_id"] = self.request.credentials.accountId
continue
if accountId != self.request.credentials.accountId:
raise VLException(
Error.ForbiddenVisibilityAreaDenied.format("handler matching policy candidates"), 403, False
)
[docs]class TasksServiceBaseHandler(BaseProxyHandler):
"""
Class for proxy handlers to tasks
"""
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a tasks service address
Returns:
a tasks service address
"""
return self.config.tasksAddress
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a tasks service timeouts
Returns:
a tasks service timeouts
"""
return self.config.tasksTimeouts
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if not self.config.additionalServicesUsage.lunaTasks:
raise VLException(Error.LunaTasksIsDisabled, statusCode=403, isCriticalError=False)
[docs] def checkTokenPermissions(self) -> None:
"""
Description see :func:`~BaseRequestHandler.checkTokenPermissions`.
"""
self.checkTokenPermissionsDefault(objectName="task")
[docs]class PythonMatcherServiceBaseHandler(BaseProxyHandler):
"""
Class for proxy handlers to python-matcher/python-matcher-proxy
"""
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a python matcher service address
Returns:
a python matcher service address
"""
return (
self.config.matcherProxyAddress
if self.config.additionalServicesUsage.lunaMatcherProxy
else self.config.pythonMatcherAddress
)
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a python matcher service timeouts
Returns:
a python matcher service timeouts
"""
return (
self.config.matcherProxyTimeouts
if self.config.additionalServicesUsage.lunaMatcherProxy
else self.config.pythonMatcherTimeouts
)
[docs]class BaseFaceSampleProxyHandler(BaseProxyHandler):
"""
Base face sample proxy.
"""
@property
def serviceAddress(self) -> ImageStoreAddressSettings:
"""
Get a image store service address
Returns:
a image store service address
"""
return self.config.faceSamplesStorage
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a image store service timeouts
Returns:
a image store service timeouts
"""
return self.config.faceSamplesStorageTimeouts
[docs]class BaseBodySampleProxyHandler(BaseProxyHandler):
"""
Base body sample proxy.
"""
@property
def serviceAddress(self) -> ImageStoreAddressSettings:
"""
Get a image store service address
Returns:
a image store service address
"""
return self.config.bodySamplesStorage
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a image store service timeouts
Returns:
a image store service timeouts
"""
return self.config.bodySamplesStorageTimeouts
[docs]class AccountsBaseHandler(BaseProxyHandler):
"""
Class for proxy handlers to accounts
"""
@property
def serviceAddress(self) -> ServiceAddressSettings:
"""
Get a faces service address
Returns:
a faces service address
"""
return self.config.accountsAddress
@property
def serviceTimeouts(self) -> ServiceTimeoutsSettings:
"""
Get a faces service timeouts
Returns:
a faces service timeouts
"""
return self.config.accountsTimeouts
[docs] def checkTokenPermissions(self) -> None:
"""
Description see :func:`~BaseRequestHandler.checkTokenPermissions`.
"""
[docs] def checkAccountType(self) -> None:
"""
Check account type in specified json is allowed
Raises:
VLException(Error.BadInputJson.format("", accountType)) if `account_type` is not allowed
"""
if self.request.json:
self.validateJson(self.request.json, schemas.OBJECT_SCHEMA)
accountType = self.request.json.get("account_type")
if accountType is not None and accountType not in ("advanced_user", "user"):
raise VLException(
Error.BadInputJson.format(
"account_type", "value is not a valid enumeration member; permitted: 'advanced_user', 'user'"
),
400,
False,
)
