Source code for luna_handlers.classes.schemas.handler

"""
Module contains schemas for handler
"""
from typing import Any, List, Optional, Union
from uuid import UUID

from pydantic import PrivateAttr, ValidationError, conlist, model_validator
from pydantic.types import StrictStr
from vlutils.structures.pydantic import UpdatableList

from classes.schemas import types
from classes.schemas.base_schema import BaseSchema, HandlerSettings
from classes.schemas.match_policy import MatchPolicy
from classes.schemas.policies import Policies
from classes.schemas.simple_schemas import BoundingBoxSchema, UrlBboxSchema, UrlWithFaceBbox
from classes.schemas.types import UUIDObject
from classes.schemas.validators import checkRequiredImageMimetype, validateOneOfStrParams
from crutches_on_wheels.cow.enums.handlers import HandlerType
from crutches_on_wheels.cow.errors.pydantic_errors import PydanticError
from crutches_on_wheels.cow.pydantic.types import CustomDatetime, OptionalNotNullable


[docs] class ValidatorHandlerModel(BaseSchema): """Handler validator model""" # handler policies policies: Policies
[docs] class CreateHandlerPolicies(Policies): _raw = PrivateAttr() def __init__(self, **data): super().__init__(**data) self._raw = data
[docs] def asDict(self): """Get data from initialized model excluding match policy on-load transformations.""" result = super().asDict() for matchPolicy, rawMatchPolicy in zip(result["match_policy"], self._raw.get("match_policy", [])): matchPolicy |= rawMatchPolicy return result
[docs] class CreateHandlerModel(BaseSchema): """Handler model for handler creation""" # handler policies policies: CreateHandlerPolicies | None = None # handler description description: types.Str128 = "" # handler type handlerType: HandlerType = HandlerType.static # lambda id lambdaId: UUID | None = None def _validateDynamicHandlerPolicies(self): """ Validate not allowed policies for dynamic handler and required policies for non-dynamic handler Args: values: values from request Raises: PydanticCustomError("ValueError", "Dynamic handler can not have any policies.") if dynamic handler flag is True and policies specified PydanticCustomError("ValueError", "Non-dynamic handler must have policies.") if dynamic handler flag is False and policies not specified PydanticCustomError("ValueError", "Lambda handler must have lambda_id.") if lambda handler flag is True and lambda id not specified PydanticCustomError("ValueError", "Lambda handler can not have any policies.") if lambda handler flag is True and policies specified """ handlerType = self.handlerType isPoliciesExists = self.policies is not None lambdaId = self.lambdaId if lambdaId is not None and handlerType != HandlerType.lambdaHandler: raise PydanticError.PydanticValidationError.format("Extra fields not allowed: 'lambda_id'")() if handlerType == HandlerType.dynamic: if isPoliciesExists is True: raise PydanticError.PydanticValidationError.format("Dynamic handler can not have any policies.")() if handlerType == HandlerType.static: if isPoliciesExists is False: raise PydanticError.PydanticValidationError.format("Non-dynamic handler must have policies.")() if handlerType == HandlerType.lambdaHandler: if lambdaId is None: raise PydanticError.PydanticValidationError.format("Lambda handler must have lambda_id.")() if isPoliciesExists is True: raise PydanticError.PydanticValidationError.format("Lambda handler can not have any policies.")()
[docs] def model_post_init(self, __context: Any): """Validate handler, post init""" self._validateDynamicHandlerPolicies() return self
[docs] class CachedHandlerModel(BaseSchema): """Cached handler model | contains minimal required fields""" # handler account id accountId: UUIDObject # handler policies policies: Optional[Policies] = None # handler type handlerType: HandlerType = HandlerType.static # lambda id lambdaId: str | None = None # handler last update time lastUpdateTime: CustomDatetime
[docs] @staticmethod def postInitMatchPolicy(matchPolicies: List[MatchPolicy], accountId: str | UUID) -> None: """ Post init matching policy - apply handler account id for matching candidates Args: matchPolicies: matching policies accountId: handler account id """ for matchPolicy in matchPolicies: matchPolicy.candidates.accountId = accountId
[docs] def model_post_init(self, __context: Any): """Post init""" if self.handlerType.value == HandlerType.dynamic.value: self.postInitMatchPolicy(self.policies.matchPolicy, self.accountId) return self
[docs] class UrlForInputEvent(UrlBboxSchema): """Url schema with detection time and image origin""" # sample detect time detectTime: CustomDatetime = OptionalNotNullable() # image origin imageOrigin: StrictStr = OptionalNotNullable() # image origin mimetype imageOriginMimetype: StrictStr = OptionalNotNullable() # user-defined timestamp relative to something, such as the start of a video detectTs: Optional[types.DetectTs] = None # event meta meta: dict = OptionalNotNullable()
[docs] class SampleForInputEvent(BaseSchema): """Sample schema""" # sample id sampleId: UUIDObject # sample detect time detectTime: CustomDatetime = OptionalNotNullable() # image origin imageOrigin: StrictStr = OptionalNotNullable() # image origin mimetype imageOriginMimetype: StrictStr = OptionalNotNullable() # user-defined timestamp relative to something, such as the start of a video detectTs: Optional[types.DetectTs] = None # event meta meta: dict = OptionalNotNullable()
UrlsWithFaceBBox = UpdatableList( UrlWithFaceBbox, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit ) UrlsWithFaceAndBodyBBox = UpdatableList( UrlBboxSchema, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit ) SamplesIDs = UpdatableList(UUIDObject, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit) UrlsForInputEvent = UpdatableList( UrlForInputEvent, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit ) SamplesForInputEvent = UpdatableList( SampleForInputEvent, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit )
[docs] class FaceInputEstimationsModel(BaseSchema): """Model for incoming face estimations: urls, image or samples""" # image (base64) image: Optional[str] = OptionalNotNullable() # image mimetype mimetype: Optional[str] = OptionalNotNullable() # face bounding box list faceBoundingBoxes: conlist(BoundingBoxSchema, min_length=1) = OptionalNotNullable() # url list urls: UrlsWithFaceBBox = OptionalNotNullable() # sample ids samples: SamplesIDs = OptionalNotNullable()
[docs] @model_validator(mode="before") def validateHandler(cls, values: dict) -> dict: """Validate handler model""" validateOneOfStrParams(values, params=["urls", "samples", "image"]) checkRequiredImageMimetype(values) return values
[docs] class HandlerInputEstimationsModel(BaseSchema): """Handler model for incoming estimations: urls, image(with face and/or body bounding boxes) or samples""" # image (base64) image: Optional[str] = OptionalNotNullable() # image mimetype mimetype: Optional[str] = OptionalNotNullable() # face bounding box list faceBoundingBoxes: conlist(BoundingBoxSchema, min_length=1) = OptionalNotNullable() # body bounding box list bodyBoundingBoxes: conlist(BoundingBoxSchema, min_length=1) = OptionalNotNullable() # list of urls with detection time urls: UrlsForInputEvent = OptionalNotNullable() # samples with ids or detection time and ids samples: Union[SamplesForInputEvent, SamplesIDs] = OptionalNotNullable() # image detection time detectTime: CustomDatetime = OptionalNotNullable() # user-defined timestamp relative to something, such as the start of a video detectTs: Optional[types.DetectTs] = None # image origin imageOrigin: StrictStr = OptionalNotNullable() # image origin mimetype imageOriginMimetype: StrictStr = OptionalNotNullable() # event meta meta: dict = OptionalNotNullable() # TODO: avoid using __init__, replace it with validators def __init__(self, **kwargs): try: super().__init__(**kwargs) except ValidationError as exc: """ Try get correct error for 'oneOf' validation for `samples` validation """ if "samples" not in kwargs or not isinstance(kwargs["samples"], list) or not kwargs["samples"]: raise exc if isinstance(kwargs["samples"][0], dict): class Model(HandlerInputEstimationsModel): def __init__(self, **kwargs): super(BaseSchema, self).__init__(**kwargs) samples: SamplesForInputEvent else: class Model(HandlerInputEstimationsModel): def __init__(self, **kwargs): super(BaseSchema, self).__init__(**kwargs) samples: SamplesIDs try: Model.model_validate(kwargs) except ValidationError as exc: raise exc
[docs] @model_validator(mode="before") def validateHandler(cls, values: dict) -> dict: """Validate handler model""" validateOneOfStrParams(values, params=["urls", "samples", "image"]) checkRequiredImageMimetype(values) return values