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 ValidationError, conlist, root_validator

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 OptionalNotNullable, userConstrainedList
from classes.schemas.validators import checkRequiredImageMimetype, validateOneOfStrParams


[docs]class ValidatorHandlerModel(BaseSchema): """Handler validator model""" # handler policies policies: Policies
[docs]class CreateHandlerModel(BaseSchema): """Handler model for handler creation""" # handler policies policies: Optional[Policies] = None # handler description description: types.Str128 = "" # handler dynamic flag isDynamic: bool = False @staticmethod def _validateDynamicHandlerPolicies(values: Any): """ Validate not allowed policies for dynamic handler and required policies for non-dynamic handler Args: values: values from request Raises: ValueError("Dynamic handler can not have any policies.") if dynamic handler flag is True and policies specified ValueError("Non-dynamic handler must have policies.") if dynamic handler flag is False and policies not specified """ isDynamic = values["isDynamic"] isPoliciesExists = values["policies"] is not None if isDynamic is isPoliciesExists is True: raise ValueError("Dynamic handler can not have any policies.") if isDynamic is isPoliciesExists is False: raise ValueError("Non-dynamic handler must have policies.")
[docs] @root_validator(pre=False, skip_on_failure=True) def validateNPostInit(cls, values): """Validate handler, post init""" cls._validateDynamicHandlerPolicies(values) return values
[docs]class CachedHandlerModel(BaseSchema): """Cached handler model | contains minimal required fields""" # handler account id accountId: UUID # handler policies policies: Optional[Policies] = None # handler dynamic flag isDynamic: bool = False # handler last update time lastUpdateTime: types.CustomDatetime
[docs] @staticmethod def postInitMatchPolicy(matchPolicies: List[MatchPolicy], accountId: str) -> 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] @root_validator(pre=False, skip_on_failure=True) def postInit(cls, values): """Post init""" if values["isDynamic"] is True: cls.postInitMatchPolicy(values["policies"].matchPolicy, values["accountId"]) return values
[docs]class UrlForInputEvent(UrlBboxSchema): """Url schema with detection time and image origin""" # sample detect time detectTime: types.CustomDatetime = OptionalNotNullable() # image origin imageOrigin: types.Str256 = OptionalNotNullable() # user-defined timestamp relative to something, such as the start of a video detectTs: Optional[types.DetectTs] = None
[docs]class SampleForInputEvent(BaseSchema): """Sample schema""" # sample id sampleId: UUID # sample detect time detectTime: types.CustomDatetime = OptionalNotNullable() # image origin imageOrigin: types.Str256 = OptionalNotNullable() # user-defined timestamp relative to something, such as the start of a video detectTs: Optional[types.DetectTs] = None
UrlsWithFaceBBox = userConstrainedList( UrlWithFaceBbox, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit ) UrlsWithFaceAndBodyBBox = userConstrainedList( UrlBboxSchema, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit ) SamplesIDs = userConstrainedList(UUID, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit) UrlsForInputEvent = userConstrainedList( UrlForInputEvent, minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit ) SamplesForInputEvent = userConstrainedList( 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] # image mimetype mimetype: Optional[str] # face bounding box list faceBoundingBoxes: conlist(BoundingBoxSchema, min_items=1) = OptionalNotNullable() # url list urls: UrlsWithFaceBBox = OptionalNotNullable() # sample ids samples: SamplesIDs = OptionalNotNullable()
[docs] @root_validator(skip_on_failure=True) 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] # image mimetype mimetype: Optional[str] # face bounding box list faceBoundingBoxes: conlist(BoundingBoxSchema, min_items=1) = OptionalNotNullable() # body bounding box list bodyBoundingBoxes: conlist(BoundingBoxSchema, min_items=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: types.CustomDatetime = OptionalNotNullable() # user-defined timestamp relative to something, such as the start of a video detectTs: Optional[types.DetectTs] = None # image origin imageOrigin: types.Str256 = OptionalNotNullable() 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.validate(kwargs) except ValidationError as exc: raise exc
[docs] @root_validator(skip_on_failure=True) def validateHandler(cls, values: dict) -> dict: """Validate handler model""" validateOneOfStrParams(values, params=["urls", "samples", "image"]) checkRequiredImageMimetype(values) return values