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 functools import cached_property
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"}


[docs]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
[docs]class BaseSchema(_BaseModel): """Base schema model"""
[docs] class Config: """Pydantic model config""" extra = Extra.forbid # additionalProperties - False arbitrary_types_allowed = True # allow user types for fields keep_untouched = (cached_property,)
[docs] @classmethod def alias_generator(cls, paramName: str) -> str: """ Generate aliases Args: paramName: model field name """ return replaceEqualSuffix(snakecase(paramName))
[docs] 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)
[docs]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
[docs] @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
[docs]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)
[docs] @classmethod def update(cls): """Update inherit schema.""" raise NotImplementedError
[docs] @classmethod def updateSchemas(cls): """Run settings update for inherited schemas.""" for obj in cls.__inheritors__: obj.update()