Source code for luna_handlers.classes.schemas.handler
"""
Module contains schemas for handler
"""
from typing import Optional, List, Any
from uuid import UUID
from pydantic import root_validator, Field, validator, conlist
from app.handlers.available_content_types import isAllowableContentType
from classes.schemas import types
from classes.schemas.base_schema import BaseSchema
from classes.schemas.match_policy import MatchPolicy
from classes.schemas.policies import Policies
from classes.schemas.simple_schemas import UrlSchema, BoundingBoxSchema
from classes.schemas.types import OptionalNotNullable
class _BaseHandlerModel(BaseSchema):
"""Handler model"""
# handler description
description: types.Str128 = ""
# handler policies
policies: Optional[Policies] = None
# handler account id
accountId: UUID
# handler dynamic flag
isDynamic: bool = False
[docs]class CreateHandlerModel(_BaseHandlerModel):
"""Handler model for handler creation"""
@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 HandlerModel(_BaseHandlerModel):
"""Handler model for handler usage"""
# handler id
handlerId: UUID
# handler creation time
createTime: types.CustomDatetime
# 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 validateNPostInit(cls, values):
"""Validate handler, post init"""
cls.postInitMatchPolicy(values["policies"].matchPolicy, values["accountId"])
return values
[docs]class BaseInputEstimationsModel(BaseSchema):
"""Base model for incoming estimations: urls, image or samples"""
# url list
urls: Optional[List[UrlSchema]] = Field(min_items=1, max_items=types.MAX_URLS_LIST_LENGTH)
# sample ids
samples: conlist(UUID, min_items=1, max_items=types.MAX_SAMPLES_LIST_LENGTH) = OptionalNotNullable()
# image (base64)
image: Optional[str]
# image mimetype (additional validator - '_mimetype')
mimetype: Optional[str]
# face bounding box list
faceBoundingBoxes: conlist(BoundingBoxSchema, min_items=1) = OptionalNotNullable()
[docs] @validator("mimetype")
def validateImageMimetype(cls, mimetype: str) -> str:
"""
Validate image mimetype
Args:
value: mimetype string
Returns:
string with valid mimetype
Raises:
ValueError(f"{value} is not allowed image mimetype") if mimetype is not allowed
"""
if not isAllowableContentType(mimetype, allowRawDescriptors=True):
raise ValueError(f"{mimetype} is not allowed image mimetype")
return mimetype
[docs] @root_validator(pre=False, skip_on_failure=True)
def validateOneOf(cls, values):
"""Validate one of"""
isRequestWithUrls = values["urls"] is not None
isRequestWithSamples = values["samples"] is not None
isRequestWithImage = values["image"] is not None
if [isRequestWithImage, isRequestWithSamples, isRequestWithUrls].count(True) != 1:
raise ValueError("Required one of 'urls', 'samples' or 'image' in request")
if isRequestWithImage:
if "mimetype" not in values:
raise ValueError("Error path: '('mimetype',)'. Error msg: 'field required''")
return values
[docs]class HandlerInputEstimationsModel(BaseInputEstimationsModel):
"""Handler model for incoming estimations: urls, image(with face and/or body bounding boxes) or samples"""
# body bounding box list
bodyBoundingBoxes: conlist(BoundingBoxSchema, min_items=1) = OptionalNotNullable()