General description =================== All considerations relevant to `lambda development <./development.html>`_ apply to Job development as well. However, since Jobs have `additional requirements <./workflow_requirements.html>`_, they also have specific development considerations. Job context =========== Unlike Lambdas, Job development may require access to different kinds of data, including information about how many requests a lambda has processed, or how much of some other work it has completed. The following structure is provided to access this data (only the part available in all Job types is described here). .. code-block:: python class LambdaContext: """Lambda context""" userCtx: UserCtx | None lambdaStartTime: float ... .. list-table:: :header-rows: 1 * - Name - Description * - userCtx - | User-defined context, if configured. | A detailed description of the user context is available in this `chapter <./development.html#user-context>`_ * - lambdaStartTime - | The time at which the Lambda started (a ``float`` value that can be compared with | the result of ``time.time()`` from the standard ``time`` module to compute the | difference with the current time). This context can be used in all functions defined to support Job execution, i.e., the functions described in the `job requirements chapter <./workflow_requirements.html>`_. The context is available for import from ``luna_lambda_tools``: .. code-block:: python from luna_lambda_tools import LambdaContext Standalone job context ---------------------- Since the Standalone Lambda is primarily intended for processing HTTP requests, its context provides, by default, information about how many requests have been processed and their outcomes when using the `Main Function <./lambda_requirements.html#lambda-general-python-code-requirements>`_, which must be defined by the user. A description of the relevant structure follows. .. code-block:: python class RequestsCount: failed: int success: int inProgress: int all: int class LambdaContext: userCtx: UserCtx | None lambdaStartTime: float previousRequestsCount: RequestsCount requestsCount: RequestsCount .. list-table:: :header-rows: 1 * - Name - Description * - requestsCount - | A structure containing data about the number of requests | processed during the current Job run. * - previousRequestsCount - | A structure containing data about the number of requests | processed during previous Job runs. Each of these structures provides the following fields: .. list-table:: :header-rows: 1 * - Name - Description * - failed - The number of failed requests. * - success - The number of successful requests. * - inProgress - The number of requests currently being processed. * - all - The total number of all requests (the sum of the fields above). Examples: In this example, the Job will terminate once the total number of successfully processed requests reaches at least 10. .. code-block:: python from luna_lambda_tools import LambdaContext async def checkCompleteness(lambdaContext: LambdaContext): if (lambdaContext.requestsCount.all + lambdaContext.previousRequestsCount.all) >= 10: return True return False In this example, the current Job run will terminate once the number of successfully processed requests reaches at least 5 during the current run, or if no requests have been processed successfully and the number of failed requests has reached at least 15. .. code-block:: python from luna_lambda_tools import LambdaContext async def checkCompleteness(lambdaContext: LambdaContext): if lambdaContext.requestsCount.success >= 5: return True if lambdaContext.requestsCount.failed >= 15 and lambdaContext.requestsCount.success == 0: return True return False .. note:: For the first run all values of all fields in `previousRequestsCount` will contain zeros. Standalone job example ---------------------- Here is an example of a Standalone Job that processes incoming requests and counts successful responses. The job completes after processing at least 2 successful requests. .. literalinclude:: examples/200/standalone_job_development/lambda/lambda_main.py :caption: lambda_main.py :language: python .. literalinclude:: examples/200/standalone_job_development/make_request.py :caption: request example :language: python Agent job context ----------------- Since the Agent Lambda is primarily designed for running video analytics on video streams, its context provides, by default, information about how many video streams have been processed, using the standard communication protocol with Luna Video Manager (see the luna-video-manager documentation for details). The relevant structure is described below. .. code-block:: python class StreamCount: count: int success: int failed: int inProgress: int class LambdaContext: userCtx: UserCtx | None lambdaStartTime: float streamCount: StreamCount previousStreamCount: StreamCount .. list-table:: :header-rows: 1 * - Name - Description * - streamCount - | A structure containing data about the number of | video streams processed during the current Job run. * - previousStreamCount - | A structure containing data about the number of | video streams processed during previous Job runs. Each of these structures provides the following fields: .. list-table:: :header-rows: 1 * - Name - Description * - failed - The number of video streams that failed to process. * - success - The number of video streams processed successfully. * - inProgress - The number of video streams currently being processed. * - count - The total number of all video streams (the sum of the fields above). Examples: In this example, the Job will terminate once the total number of successfully processed video streams reaches at least 10. .. code-block:: python from luna_lambda_tools import LambdaContext async def checkCompleteness(lambdaContext: LambdaContext): if (lambdaContext.streamCount.count + lambdaContext.previousStreamCount.count) >= 10: return True return False In this example, the current Job run will terminate once the number of successfully processed video streams reaches at least 5 during the current run, or if no video streams have been processed successfully and the number of failed video streams has reached at least 15. .. code-block:: python from luna_lambda_tools import LambdaContext async def checkCompleteness(lambdaContext: LambdaContext): if lambdaContext.streamCount.success >= 5: return True if lambdaContext.streamCount.failed >= 15 and lambdaContext.streamCount.success == 0: return True return False .. note:: For the first run all values of all fields in `previousStreamCount` will contain zeros. Agent job example ----------------- .. note:: A video analytics development description presented in `this chapter `_. Examples of lambda video-agentы from `lambda agent examples <./agent.html#agent-lambda-examples>`_ actual for jobs except for the absence of job mechanics. Here is an example of an Agent Job that processes video streams using suit analytics. The Job will terminate after successfully processing 1 video stream. .. code-block:: :caption: Lambda agent with suit analytics example file structure (the case when analytics includes in lambda as package) ├──pyproject.toml ├──poetry.lock └──lambda_main.py The `lambda_main.py` module: .. literalinclude:: examples/200/agent_job_video_analytics/lambda/lambda_main.py :caption: lambda_main.py :language: python It this case, the dependencies of lambda agent must include only analytics as dependency, not analytics dependencies: .. raw:: html
*pyproject.toml* .. literalinclude:: examples/200/agent_lambda_suit/lambda/pyproject_minimal.toml :caption: pyproject.toml :language: toml .. raw:: html
| The script demonstrates how to run a job-agent that processes video streams using suit analytics: .. literalinclude:: examples/200/agent_job_video_analytics/make_request.py :caption: request example :language: python Resource-constrained sequential processing ------------------------------------------ When computational resources are limited — for example, it needs to process **N** analytics across a large set of video, but you only have resources to run **1–2 agents** simultaneously — you can implement a sequential processing pattern. This approach creates **N workflow instances** (one agent job per analytics), and jobs will process them one by one, ensuring all videos are covered by all analytics without requiring simultaneous agent execution. .. note:: Because stream creation is not available until all required analytics registered and analytics registration is only can be performed by agent (for more information see *Luna-Video-Manager* documentation), it is allows to set *service_launch* at *deploy_parameters* within which the one *service* job will be created (if it is agent job it will perform its analytics registration) and immidiately shut down. The idea is as follows: 1. Create **N workflows**, each containing a specific analytics configuration. 2. Use ``checkCompletenessExecution`` to signal that the current job should finish after successfully processing a limited number of videos (e.g., 3 videos). 3. Use ``checkCompleteness`` that always returns ``False`` to prevent early termination within a job run. With this setup, a job starts, processes 3 videos from the first analytics workflow, then terminates. The next job launches with the second analytics, processes another 3 videos, and so on. This continues until all videos are processed by all analytics — not simultaneously, but sequentially. .. code-block:: python from luna_lambda_tools import LambdaContext async def checkCompletenessExecution(lambdaContext: LambdaContext) -> bool: """ Mark the job as complete after successfully processing 3 video videos in the current run. """ if lambdaContext.streamCount.success >= 3: return True return False async def checkCompleteness(lambdaContext: LambdaContext) -> bool: """ Always return False to prevent early termination within a job run. The job will only finish when checkCompletenessExecution returns True. """ return False In this scenario: - **N workflows** are created, each with a different analytics configuration. - A job (or several jobs, depending on ``parallel_job_limit``) is launched. - Each job processes 3 videos, then terminates. - The next job picks up the next analytics batch and processes another 3 videos. - This continues until all N analytics have processed all videos. The result is that **all videos are processed for all analytics**, just sequentially rather than in parallel. This pattern is ideal for environments with constrained resources where you cannot afford to run multiple agents simultaneously. **Enhanced example with total count tracking and busy protection:** To improve this pattern, you can add a maximum video count threshold in ``checkCompleteness`` to stop all processing once the total number of processed videos reaches a target. Use ``previousStreamCount.count + streamCount.count`` to get the cumulative count across all job runs. Additionally, use ``checkBusy`` to prevent the job from being terminated while videos are still being processed (when ``streamCount.inProgress > 0``). .. code-block:: python from luna_lambda_tools import LambdaContext # Total number of videos to process across all analytics TOTAL_VIDEOS_TO_PROCESS = 100 async def checkCompletenessExecution(lambdaContext: LambdaContext) -> bool: """ Mark the job as complete after successfully processing 3 video videos in the current run. """ if lambdaContext.streamCount.success >= 3: return True return False async def checkCompleteness(lambdaContext: LambdaContext) -> bool: """ Stop all processing once the total number of processed videos reaches the target across all job runs. """ total_processed = ( lambdaContext.streamCount.count + lambdaContext.previousStreamCount.count ) if total_processed >= TOTAL_VIDEOS_TO_PROCESS: return True return False async def checkBusy(lambdaContext: LambdaContext) -> bool: """ Keep the job alive while videos are still being processed. """ if lambdaContext.streamCount.inProgress > 0: return True return False Job helpers =========== Additionally, a set of helper functions is available for use in all the above-described functions or anywhere else within a Job. These functions help determine the current state of the Job at any given time. getRemainingJobTime ------------------- This function returns the remaining time, in seconds, until the Luna Lambda service terminates the Job forcibly. .. code-block:: python from luna_lambda_tools import getRemainingJobTime print(getRemainingJobTime()) # 123 Example: using this function to check whether the Job has sufficient remaining time. In this case, the Job will start a long-running background task on startup and makes job busy until background task is done. .. code-block:: python import asyncio from luna_lambda_tools import LambdaContext, getRemainingJobTime async def startLongBackgroundTask(): """Just an example""" await asyncio.sleep(1000) class UserCtx: """ Custom lambda context """ def __init__(self): self.task = None async def onStart(self): self.task = asyncio.create_task(startLongBackgroundTask) async def checkBusy(lambdaContext: LambdaContext) -> bool: return lambdaContext.userCtx.task is not None and lambdaContext.userCtx.task.done() getJobWorkTime -------------- This function returns the elapsed time, in seconds, since the Job started. .. code-block:: python from luna_lambda_tools import getJobWorkTime print(getJobWorkTime()) # 456 Example: using this function so that the Job starts accepting requests only 50 seconds after its start. .. code-block:: python from luna_lambda_tools import getJobWorkTime, StandaloneLambdaRequest async def main(request: StandaloneLambdaRequest): if getJobWorkTime() < 50: return {"status": "not ready"} return {"status": "ready"}