Source code for luna_lambda_tools.public.tasks

import abc

from ..private.request import ClientsKwargsMixin
from ..private.tasks import _BaseLambdaTask
from .clients import Clients
from .clients.events import Events
from .clients.faces import Faces
from .clients.lis import LIS
from .clients.lpm import LPM
from .clients.lrs import LRS
from .clients.sender import Sender




class BaseLambdaTask(_BaseLambdaTask, ClientsKwargsMixin):
    """Lambda task"""



    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)


    @property
    def clients(self):
        return Clients(
            faces=Faces(**self._getFacesClientKwargs()),
            events=Events(**self._getEventsClientKwargs()),
            matcher=LPM(**self._getLPMClientKwargs()),
            faceSamplesStore=LIS(**self._getFacesSamplesClientKwargs()),
            bodySamplesStore=LIS(**self._getBodySamplesClientKwargs()),
            imageOriginStore=LIS(**self._getImageOriginClientKwargs()),
            sdk=LRS(**self._getLRSClientKwargs()),
            sender=Sender(**self._getSenderClientKwargs()),
        )

    @abc.abstractmethod
    async def splitTasksContent(self, content: dict) -> list[dict]:
        """Split task content to sub task contents"""
        ...

    @abc.abstractmethod
    async def executeSubtask(self, subTaskContent: dict) -> dict | list:
        """Execute current sub task processing"""
        ...