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