Source code for luna_handlers.classes.schemas.base_schema

"""
Module contains base schema for other complex pydantic models and simple schemas to use in other places
"""
from abc import abstractmethod, ABCMeta
from datetime import datetime
from typing import Any, ClassVar, no_type_check
from uuid import UUID

from pydantic import BaseModel as _BaseModel, Extra
from pydantic.main import ModelMetaclass
from stringcase import snakecase
from vlutils.helpers import convertTimeToString

from classes.functions import NoneLessDict

SUFFIX_MAP = {"_gte": "__gte", "_lte": "__lte", "_gt": "__gt", "_lt": "__lt"}


def replaceEqualSuffix(snakeCaseName: str) -> str:
    """
    Replace gt/gte/lt/lte suffix contains one underline with two underlines
    Args:
        snakeCaseName: key in snake_case
    """
    for oneUnderLineKey, twoUnderLineKey in SUFFIX_MAP.items():
        if snakeCaseName.endswith(oneUnderLineKey):
            snakeCaseName = snakeCaseName.replace(oneUnderLineKey, twoUnderLineKey)
            break
    return snakeCaseName


class BaseSchema(_BaseModel):
    """Base schema model"""

    class Config:
        """Pydantic model config"""

        extra = Extra.forbid  # additionalProperties - False
        arbitrary_types_allowed = True  # allow user types for fields

        @classmethod
        def alias_generator(cls, paramName: str) -> str:
            """
            Generate aliases
            Args:
                paramName: model field name
            """
            return replaceEqualSuffix(snakecase(paramName))

    def asDict(self) -> dict:
        """
        Get data from initialized model
        Returns:
            dict with snake_case keys
        """

        def getValue(value_: Any) -> Any:
            """Convert uuid, datetime to string"""
            if isinstance(value_, UUID):
                return str(value_)
            if isinstance(value_, datetime):
                return convertTimeToString(value_)
            return value_

        res = dict()
        for key, value in self.__dict__.items():
            snakeCaseKey = replaceEqualSuffix(snakecase(key))
            if isinstance(value, BaseSchema):
                res[snakeCaseKey] = value.asDict()
            elif isinstance(value, list):
                res[snakeCaseKey] = [row.asDict() if isinstance(row, BaseSchema) else getValue(row) for row in value]
            else:
                res[snakeCaseKey] = getValue(value)

        return NoneLessDict(**res)


class HandlerSettings:
    """Container class for service settings."""

    faceDescriptorVersion: int = 0
    bodyDescriptorVersion: int = 0
    livenessRealThreshold: float = 0.0
    receivedImagesLimit: int = 1
    rawEventDetectionsLimit: int = 1
    rawEventArraysLimit: int = 1
    resultCandidateLimit: int = 1
    defaultAttributeTTL: int = 1
    maxAttributeTTL: int = 1

    @classmethod
    def initialize(
        cls,
        faceDescriptorVersion: int,
        bodyDescriptorVersion: int,
        livenessRealThreshold: float,
        handlersLimitsSettings: "HandlersLimitsSettings",
        attributeStorageSettings: "RedisKeyStorageSetting",
    ):
        """
        Initialize settings for schemas.

        Args:
            faceDescriptorVersion: default face descriptor version
            bodyDescriptorVersion: default body descriptor version
            livenessRealThreshold: default liveness threshold setting
            handlersLimitsSettings: luna handlers limits settings
            attributeStorageSettings: attribute storage settings
        """
        cls.faceDescriptorVersion = faceDescriptorVersion
        cls.bodyDescriptorVersion = bodyDescriptorVersion
        cls.livenessRealThreshold = livenessRealThreshold
        cls.receivedImagesLimit = handlersLimitsSettings.receivedImagesLimit
        cls.rawEventDetectionsLimit = handlersLimitsSettings.rawEventDetectionsLimit
        cls.rawEventArraysLimit = handlersLimitsSettings.rawEventArraysLimit
        cls.resultCandidateLimit = handlersLimitsSettings.resultCandidateLimit
        cls.defaultAttributeTTL = attributeStorageSettings.defaultTTL
        cls.maxAttributeTTL = attributeStorageSettings.maxTTL


class ExtendedModelMeta(ModelMetaclass):
    """
    Metaclass for `SchemaUpdater`, add a list of inherited classes from the current one.
    Needed to recursively call schema updates.
    """

    __inheritors__ = set()

    @no_type_check
    def __new__(mcs, name, bases, namespace, **kwargs) -> ABCMeta:
        klass = super().__new__(mcs, name, bases, namespace, **kwargs)
        for base in klass.mro():
            if isinstance(base, ExtendedModelMeta) and base.__name__ != "SchemaUpdaterMixin":
                mcs.__inheritors__.add(klass)
        return klass


class SchemaUpdaterMixin(_BaseModel, metaclass=ExtendedModelMeta):
    """Extended base schema model."""

    # handler configs, initialized from application
    settings: ClassVar[HandlerSettings] = HandlerSettings

    @classmethod
    @abstractmethod
    def update(cls):
        """Update inherit schema."""

    @classmethod
    def updateSchemas(cls):
        """Run settings update for inherited schemas."""

        for klass in cls.__inheritors__:
            klass.update()