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