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"}


[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
[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()