Source code for luna_backport3.app.handlers.base_handler

# -*- coding: utf-8 -*-
""" Base handler

Module realize base class for all handlers.
"""
from luna3.accounts.http_objs import TokenPermissions
from luna3.common.exceptions import LunaApiException
from luna3.image_store.image_store import StoreApi
from luna3.lunavl.httpclient import LunaHttpClient
from sanic.response import HTTPResponse

from app.app import Backport3App
from app.handlers.base_request import Backport3Request
from configs.configs.configs.services import SettingsBackport3
from configs.configs.configs.settings.classes import 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 VLBaseProxyHandler
from crutches_on_wheels.cow.web.handlers import BaseHandler as VLBaseHandler
from db.db_context import DBContext

TOKEN_PERMISSIONS = TokenPermissions(
    handler=["creation"],
    token=["creation", "view", "modification", "deletion"],
    attribute=["creation", "view", "modification", "deletion", "matching"],
    face=["creation", "view", "modification", "deletion", "matching"],
    list=["creation", "view", "modification", "deletion"],
    event=["creation", "view", "matching"],
    faceSample=["creation", "view", "deletion"],
    image=["creation", "view", "deletion"],
    resources=["liveness"],
)


class BaseHandler(VLBaseHandler):
    """
    Base handler for other handlers.
    """

    # backport3 input request (typing for ide)
    request: Backport3Request

    def __init__(self, request: Backport3Request):
        super().__init__(request)
        self.accountId = request.accountId
        self.addDataForMonitoring(tags={"account_id": self.accountId})
        self.lunaApiClient: LunaHttpClient = request.apiClient
        self.portraitsStoreClient: StoreApi = request.portraitsClient
        self.dbContext: DBContext = request.dbContext

    async def checkDescriptor(self, descriptorId: str) -> bool:
        """
        Check descriptor existence
        Args:
            descriptorId: descriptor id
        Returns:
            True if descriptor exists otherwise False
        Raises:
            VLException(statusCode=500, error=Error.UnknownServiceError) if check face is failed
        """
        try:
            await self.lunaApiClient.checkFace(faceId=descriptorId, raiseError=True)
        except LunaApiException as exc:
            if exc.statusCode == 404:
                return False
            if exc.body:
                raise
            raise VLException(
                Error.UnknownServiceError.format(
                    "luna-faces", "HEAD", f"{self.lunaApiClient.baseUri}/faces/{descriptorId}"
                ),
                statusCode=500,
                isCriticalError=False,
            )
        return True

    async def options(self, **kwargs) -> HTTPResponse:
        """
        Method "OPTIONS"
        """
        allowHeaders = {"Allow": ", ".join(self.getAllowedMethods())}
        return self.success(body=None, extraHeaders=allowHeaders)

    @property
    def app(self) -> Backport3App:
        """
        Get running app
        Returns:
            app
        """
        return self.request.app

    @property
    def config(self) -> SettingsBackport3:
        """
        Get app config
        Returns:
            app config
        """
        return self.app.ctx.serviceConfig


class APIProxyBaseHandler(BaseHandler, VLBaseProxyHandler):
    """
    Class for proxy handlers to luna-api
    """

    @property
    def serviceAddress(self) -> ServiceAddressSettings:
        """
        Get a image store service address
        Returns:
            a image store service address
        """
        return self.config.apiAddress

    @property
    def serviceTimeouts(self) -> ServiceTimeoutsSettings:
        """
        Get a image store service timeouts
        Returns:
            a image store service timeouts
        """
        return self.config.apiTimeouts

    def prepareHeaders(self) -> dict[str, str]:
        """
        Extends headers with auth
        """
        headers = super().prepareHeaders()
        headers.update(self.request.apiClient.getAuthHeader())
        return headers