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 Int01, 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()
# is vl detections or not
trustedDetections: Int01 = 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()
# is vl detections or not
trustedDetections: Int01 = 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()
# is vl detections or not
trustedDetections: Int01 = 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()
# is vl detections or not
trustedDetections: Int01 = 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