General description

All considerations relevant to lambda development apply to Job development as well. However, since Jobs have additional requirements, 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).

class LambdaContext:
    """Lambda context"""
    userCtx: UserCtx | None
    lambdaStartTime: float
    ...

Name	Description
userCtx	User-defined context, if configured. A detailed description of the user context is available in this chapter
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.

The context is available for import from luna_lambda_tools:

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, which must be defined by the user. A description of the relevant structure follows.

class RequestsCount:
    failed: int
    success: int
    inProgress: int
    all: int

class LambdaContext:
    userCtx: UserCtx | None
    lambdaStartTime: float
    previousRequestsCount: RequestsCount
    requestsCount: RequestsCount

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:

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.

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.

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.

lambda_main.py

from luna_lambda_tools import LambdaContext, StandaloneLambdaRequest, logger


class UserCtx:
    async def onStart(self):
        logger.info("Job started")

    async def onShutdown(self):
        logger.info("Job shutdown")


async def checkCompleteness(lambdaContext: LambdaContext) -> bool:
    """
    Job completes after processing at least 2 successful requests
    (across all runs, including previous runs).
    """
    total = lambdaContext.requestsCount.success + lambdaContext.previousRequestsCount.success
    return total >= 2


async def main(request: StandaloneLambdaRequest) -> dict:
    """
    Processes incoming requests and counts successful responses.

    Supposed request structure:

    ```json
    {"message": "hello"}
    ```

    """
    logger.info("Processing request: %s", request.json)
    message = request.json.get("message", "no message")
    result = f"Processed: {message}"
    logger.info("Request processed: %s", result)
    return {"status": "ok", "result": result}

request example

from time import sleep, time

from luna3.luna_lambda.luna_lambda import LambdaApi

SERVER_ORIGIN = "http://lambda_address:lambda_port"  # Replace by your values before start
SERVER_API_VERSION = 2
lambdaApi = LambdaApi(origin=SERVER_ORIGIN, api=SERVER_API_VERSION)
workflowId, accountId = "your_workflow_id", "your_account_id"  # Replace by your values before start


def waitJobDone(workflowId: str, jobId: str, timeout: int = 60) -> None:
    """
    Wait until job status changes from 'running' to 'completed' or 'error'.
    Raises TimeoutError if job does not complete within timeout seconds.
    """
    end_time = time() + timeout
    while time() < end_time:
        reply = lambdaApi.getJob(workflowId=workflowId, jobId=jobId, accountId=accountId, raiseError=True)
        status = reply.json["info"].get("last_execution_status", None)
        if status in ("completed", "error"):
            return
        sleep(2)
    raise TimeoutError(f"Job did not complete within {timeout} seconds")


def makeRequest():
    """
    Send requests to a running job via proxy and wait for job completion.

    Before running this script:
    1. Create a workflow from the lambda archive (createWorkflow API)
    2. Create a job within the workflow (createJob API)
    3. Wait for the job to start running (job status should be 'running')

    Then send 2 requests to the job via proxy. The job will complete
    after processing 2 successful requests (see checkCompleteness in lambda_main.py).
    After sending requests, waits for job to finish (within 60 seconds).
    """
    jobs = lambdaApi.getJobs(workflowId=workflowId, accountId=accountId, raiseError=True).json["jobs"]
    if len(jobs) != 1:
        raise ValueError("Expected exactly 1 job")
    jobId = jobs[0]["job_id"]
    for i in range(2):
        lambdaApi.proxyJobPost(
            workflowId=workflowId,
            jobId=jobId,
            path="main",
            accountId=accountId,
            body={"message": f"request #{i+1}"},
        )
        sleep(1)

    waitJobDone(workflowId=workflowId, jobId=jobId, timeout=60)

    # Get final job status
    finalReply = lambdaApi.getJob(workflowId=workflowId, jobId=jobId, accountId=accountId, raiseError=True)
    return finalReply


if __name__ == "__main__":
    response = makeRequest()
    print(response.json)

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.

class StreamCount:
    count: int
    success: int
    failed: int
    inProgress: int

class LambdaContext:
    userCtx: UserCtx | None
    lambdaStartTime: float
    streamCount: StreamCount
    previousStreamCount: StreamCount

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:

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.

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.

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 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.

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:

lambda_main.py

"""
Job Agent with Video Analytics Example

This lambda is designed to run as a Job in a Workflow.
It processes video streams using SUIT analytics (object detection via ResNet-50).

The Job will terminate after successfully processing 1 video stream.

How to run:
1. Create a workflow from this lambda archive (createWorkflow API)
2. Create a job within the workflow (createJob API)
3. The job will start processing video streams automatically
4. The job terminates when checkCompleteness() returns True

See: docs/sphinx/source/workflow_development.rst for details on job development.
"""

# Available analytics modules to use
AVAILABLE_ANALYTICS = ["analytics_suit"]

# Whether to use FSDK
USE_FSDK = True


async def checkCompleteness(lambdaContext) -> bool:
    """
    Check if the job should terminate.

    The job terminates after successfully processing at least 1 video stream.
    This function is called by the Luna Lambda service to determine if the job
    has completed its work and can be safely terminated.

    Args:
        lambdaContext: LambdaContext from luna_lambda_tools containing
            streamCount with information about processed video streams.

    Returns:
        True if the job should terminate, False otherwise.
    """

    # Check if we have successfully processed at least 1 stream
    # during the current run OR previous runs
    total_success = lambdaContext.streamCount.success + lambdaContext.previousStreamCount.success
    if total_success >= 1:
        return True

    return False

It this case, the dependencies of lambda agent must include only analytics as dependency, not analytics dependencies:

*pyproject.toml*

pyproject.toml

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

[tool.poetry]
name = "lambda-agent"
version = "0.0.1"
description = "lambda-agent"
authors = [ "VisionLabs", ]

[[tool.poetry.source]]
name = "vlabspypi"
url = "http://pypi.visionlabs.ru/root/public/+simple"
priority = "primary"

[[tool.poetry.source]]
name = "public-pypi"
url = "https://pypi.org/simple"
priority = "supplemental"

[tool.poetry.dependencies]
python = "^3.12"
suit-analytics = "0.0.4"

The script demonstrates how to run a job-agent that processes video streams using suit analytics:

request example

"""
Job Agent Video Analytics - Test Script

This script demonstrates how to run a job-agent that processes video streams
using SUIT analytics.

How to run:
1. Create a workflow from the lambda archive (createWorkflow API)
2. Create a job within the workflow (createJob API)
3. The job will start processing video streams automatically
4. Wait for the job to complete (status changes to 'done' or 'error')

Before running:
- Replace SERVER_ORIGIN with your lambda service address
- Replace VIDEO_MANAGER_ORIGIN and EVENTS_ORIGIN with your service addresses
"""

from time import sleep, time

from luna3.events.events import EventsApi
from luna3.luna_lambda.luna_lambda import LambdaApi
from luna3.video_manager.http_objs import StreamAnalytic
from luna3.video_manager.video_manager import VideoManagerApi

SERVER_ORIGIN = "http://lambda_address:lambda_port"  # Replace with your lambda address and port
SERVER_API_VERSION = 2
lambdaApi = LambdaApi(origin=SERVER_ORIGIN, api=SERVER_API_VERSION)
workflowId, accountId = "your_workflow_id", "your_account_id"  # Replace by your values before start
videoManagerApi = VideoManagerApi(origin="video_manager_origin", api="video_manager_api_version")
eventsApi = EventsApi(origin="events_origin", api="events_api_version")


def waitStreamIsDone(streamId):
    # wait stream processing is done | 60 sec timeout
    st = time()
    while time() - st < 60:
        if videoManagerApi.getStream(streamId=streamId).json["status"] == 2:
            return
        sleep(0.05)
    raise TimeoutError("Failed to wait for stream processing is done")


def waitJobDone(workflowId: str, jobId: str, timeout: int = 60) -> None:
    """
    Wait until job status changes from 'running' to 'completed' or 'error'.
    Raises TimeoutError if job does not complete within timeout seconds.
    """
    end_time = time() + timeout
    while time() < end_time:
        reply = lambdaApi.getJob(workflowId=workflowId, jobId=jobId, accountId=accountId, raiseError=True)
        status = reply.json["info"].get("last_execution_status", None)
        if status in ("completed", "error"):
            return
        sleep(2)
    raise TimeoutError(f"Job did not complete within {timeout} seconds")


def makeRequest():
    """
    Create a workflow, create a job, start a video stream, and wait for completion.

    Returns:
        The job status response.
    """
    jobs = lambdaApi.getJobs(workflowId=workflowId, accountId=accountId, raiseError=True).json["jobs"]
    if len(jobs) != 1:
        raise ValueError("Expected exactly 1 job")
    jobId = jobs[0]["job_id"]

    # Create a video stream with analytics
    streamId = videoManagerApi.createStream(
        accountId=accountId,
        streamType="videofile",
        reference="your_video_url",  # Replace with actual video URL
        analytics=[
            StreamAnalytic(
                name="analytics_suit",
                parameters={
                    "parameters": {"rate": {"period": 0.5, "unit": "second"}},
                    "callbacks": [{"type": "luna-event"}],
                },
            )
        ],
        raiseError=True,
    ).json["stream_id"]

    # Wait for stream processing to complete
    waitStreamIsDone(streamId=streamId)

    # Get events created during stream processing
    reply = eventsApi.getGeneralEvents(streamIds=[streamId])

    waitJobDone(workflowId=workflowId, jobId=jobId, timeout=60)

    return reply


if __name__ == "__main__":
    response = makeRequest()
    print(response.text)

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.

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

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.

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.

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.

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.

from luna_lambda_tools import getJobWorkTime, StandaloneLambdaRequest

async def main(request: StandaloneLambdaRequest):
    if getJobWorkTime() < 50:
        return {"status": "not ready"}
    return {"status": "ready"}