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