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 datetime import datetime
from typing import Any, ClassVar
from uuid import UUID

from pydantic import BaseModel as _BaseModel, Extra
from stringcase import snakecase

from configs.configs.configs.settings.classes import HandlersLimitsSettings, RedisKeyStorageSetting
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 hasattr(value, "asDict"):
                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 SchemaUpdaterMixin:
    """Extended base schema model."""

    # handler configs, initialized from application
    settings: ClassVar[HandlerSettings] = HandlerSettings
    # list of inherited classes from the current one
    __inheritors__: list = []

    def __init_subclass__(cls, **kwargs):
        cls.__inheritors__.append(cls)

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

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

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