General description

Independently of lambda type there are several general things whose applicable for every lambda.

Luna lambda tools

It is required the luna-lambda-tools package which is available at VL pypi for development. This package is available as luna_lambda_tools library and will be used in all examples below.

To install luna-lambda-tools locally, execute:

pip install --trusted-host pypi.visionlabs.ru --extra-index-url http://pypi.visionlabs.ru/root/public/+simple luna-lambda-tools

The luna-lambda-tools package decided in two parts - public and private: private part is not intended for user and does not guarantee backward compatibility, public part is intended for user and guarantees backward compatibility.

The public part code description.

luna_lambda_tools.public

User context

It is possible to make the lambda do some stuff after startup and before shutdown declaring UserCtx class with asynchronous onStart and/or onShutdown functions.

For example, it is possible to initialize and close connection to database:

from luna_lambda_tools import logger

def setupDB():
    ...

def closeDBConnections():
    ...

class UserCtx:
    async def onStart(self):
        logger.info("start lambda")
        setupDB()

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

It is possible to use initialize UserCtx at application start, use during request processing and close at application shutdown:

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest


class UserCtx:
    def __init__(self):
        self.value = 0

    async def onStart(self):
        self.value = 1

    async def onShutdown(self):
        self.value = 0

    def getValue(self):
        return self.value


async def main(request: StandaloneLambdaRequest) -> dict:
    return {"result": request.userCtx.getValue()}

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId)
    return reply


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

This example demonstrates making requests to external server using for session for all requests:

lambda_main.py

import aiohttp
from luna_lambda_tools import StandaloneLambdaRequest, logger


class UserCtx:
    def __init__(self):
        self.session = None

    async def onStart(self):
        logger.info("start lambda")
        self.session = aiohttp.ClientSession()

    async def onShutdown(self):
        logger.info("shutdown lambda")
        await self.session.close()


async def main(request: StandaloneLambdaRequest) -> dict:
    url = "https://docs.visionlabs.ai/luna"
    async with request.userCtx.session.get(url) as response:
        statusCode = response.status
    return {"status": statusCode}

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId)
    return reply


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

Logger

It is possible to use logger from luna-lambda-tools anywhere in lambda.

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest, logger


async def main(request: StandaloneLambdaRequest) -> dict:
    logger.info("Hello, world!")
    return {"result": "empty"}

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId)
    return reply


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

Lambda modules structure

It is possible to add and use more than one python module. Here is an example of file structure and python source code which is required for proper lambda work.

Archive file structure

 ├──lambda_main.py
 ├──path
 │  ├──fileinpath1.py
 │  └──fileinpath2.py
 └──file1.py

lambda_main.py

from file1 import foo_from_file1
from luna_lambda_tools import StandaloneLambdaRequest
from path import fileinpath1
from path.fileinpath2 import foo_from_fileinpath2


async def main(request: StandaloneLambdaRequest) -> dict:
    return {
        "from_file_1": foo_from_file1(),
        "variable_from_nested": fileinpath1.get_variable_from_another_file(),
        "from_file_in_path_1": fileinpath1.foo_from_fileinpath1(),
        "from_file_in_path_2": foo_from_fileinpath2(),
    }

file1.py

def foo_from_file1():
    return "foo_from_file1"

fileinpath1.py

from .fileinpath2 import variable

def foo_from_fileinpath1():
    return "foo_from_fileinpath1"

def get_variable_from_another_file():
    return variable

fileinpath2.py

variable = 1


def foo_from_fileinpath2():
    return "foo_from_fileinpath2"

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId)
    return reply


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

User configuration file

It is possible to add a configuration file for lambda in YAML format. The file must be named lambda_config.yml and placed in the root of the zip archive.

Archive file structure

├──lambda_main.py
└──lambda_config.yml

You can import this configuration to your lambda code and use as python dictionary object.

lambda_config.yml

name: "My Lambda"
version: 5
database:
  type: "postgresql"
  host: "10.8.0.1"
  port: 5432
  name: "my_lambda_db"
  user: "db_user"
  password: "db_password"
logging:
  level: "info"
  format: "text"
features:
  feature_1: true
  feature_2: false
users:
- name: "admin"
  email: "admin@example.com"
- name: "user"
  email: "user@example.com"

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest, lambdaConfig


async def main(request: StandaloneLambdaRequest) -> dict:
    result = {
        "title": lambdaConfig["name"],
        "version": lambdaConfig["version"],
        "feature1_enabled": lambdaConfig["features"]["feature_1"],
        "username": lambdaConfig["users"][0]["name"],
    }
    return result

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId)
    return reply


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

Lambda exceptions

It is possible to separate exceptions into two types: expected and unexpected. Expected exceptions must be inherited from UserException. For example:

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest, UserException


class ImageCountException(UserException):
    statusCode = 400
    errorText = "expected two images in request"


async def main(request: StandaloneLambdaRequest) -> dict:
    if len(request.json["images"]) != 2:
        raise ImageCountException
    return {"image_count": len(request.json["images"])}

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    data = {"images": ["image_one"]}
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId, body=data)
    return reply


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

In the case presented above for any request with images quantity other than two, it will raise ImageCountException and then, luna-handlers will process exception and return a response to the user with 400 status code and expected two images in request message in detail.

Unexpected exceptions will proceed in another way. If any exception occurs, luna-handlers will return a response to the user with 500 status code and exception text in detail. For example:

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest

async def main(request: StandaloneLambdaRequest) -> dict:
   abc = 1/0
   ...

will cause response with division by zero as exception detail.

Luna services clients

The lambda can use luna-services for different operations.

The clients from luna-lambda-tools represent wrap for VL luna3 library (also available at VL pypi). As well as luna-lambda-tools, luna3 has a public part that is intended for user and guarantee backward compatibility.

Example of clients usage for standalone/handlers lambdas (for clients usage in tasks lambdas see tasks lambda examples):

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest


async def main(request: StandaloneLambdaRequest) -> dict:
    versions = {
        "faces_address": request.clients.faces.getAddress(),
    }
    if request.eventsEnabled:
        versions["events_address"] = request.clients.events.getAddress()
    return versions

request example

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path="main", accountId=accountId)
    return reply


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

Luna services clients description

It also possible to check whether luna service enabled or disabled using several properties:

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest

async def main(request: StandaloneLambdaRequest) -> dict:
    request.logger.info(f"events enabled: {request.eventsEnabled}")
    request.logger.info(f"sender enabled: {request.senderEnabled}")
    request.logger.info(f"handlers enabled: {request.handlersEnabled}")
    ...

It can be useful if lambda can optional use one of this service, for example, save event if Luna-Events is enabled:

lambda_main.py

from luna_lambda_tools import StandaloneLambdaRequest

async def main(request: StandaloneLambdaRequest) -> dict:
    if request.eventsEnabled:
        request.clients.events.saveEvents(events=[...])
    return {"status": "success"}

Additional routes

Each lambda type can be extended with additional routes with custom functions. For example:

standalone lambda with additional routes. Additional routes for handlers lambda can be added the same way.

lambda_main.py

from luna_lambda_tools import LambdaUserHandler, StandaloneLambdaRequest

UUID_REGEXP_STR = r"[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}"

PARAMETERS = []


class ParameterHandler(LambdaUserHandler):
    """Parameter handler"""

    route: str = f"/parameters/<parameter:{UUID_REGEXP_STR}>"

    async def post(self, parameter):
        PARAMETERS.append(parameter)
        return self.sendResponse(201, json={"response": parameter})

    async def head(self, parameter):
        if parameter not in PARAMETERS:
            return self.sendResponse(404, body="Parameter not found", headers={"Content-Type": "text/plain"})
        return self.sendResponse(204)


class ParametersHandler(LambdaUserHandler):
    """Parameters handler"""

    route: str = "/parameters"

    async def get(self):
        return self.sendResponse(200, body="; ".join(PARAMETERS), headers={"Content-Type": "text/plain"})


ROUTES_TO_ADD = [ParameterHandler, ParametersHandler]


async def main(request: StandaloneLambdaRequest) -> dict:
    """Main route"""
    return {"result": "main route response"}

request example

from uuid import uuid4

from luna3.luna_lambda.luna_lambda import LambdaApi

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


def makeRequest():
    reply = lambdaApi.proxyLambdaPost(lambdaId=lambdaId, path=f"parameters/{str(uuid4())}", accountId=accountId)
    return reply


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

tasks lambda with additional routes.

lambda_main.py

import asyncio
from collections import defaultdict

from luna_lambda_tools import LambdaUserHandler
from luna_lambda_tools.public.tasks import BaseLambdaTask

SUBTASKS_DONE = defaultdict(lambda: 0)


class DoneSubtasksHandler(LambdaUserHandler):
    """Subtasks handler"""

    route: str = r"/subtasks_done/<taskId:\d+>"

    async def get(self, taskId):
        """Get subtasks done"""
        if (count := SUBTASKS_DONE.get(int(taskId), None)) is None:
            return self.sendResponse(404, json={"error": f"Task {taskId} not found"})
        return self.sendResponse(200, json={"sub_tasks_done": count})


ROUTES_TO_ADD = [DoneSubtasksHandler]


class LambdaTask(BaseLambdaTask):
    """Lambda task"""

    async def splitTasksContent(self, content: dict) -> list[dict]:
        """Split task content to sub task contents"""
        return [{"content": data} for data in content["data"]]

    async def executeSubtask(self, subtaskContent: dict) -> dict | list:
        """Execute current sub task processing"""
        for _ in range(subtaskContent["content"]["count"]):
            await asyncio.sleep(subtaskContent["content"]["delay"])

        SUBTASKS_DONE[self._data.taskId] += 1
        return {"result": "some task result"}

Each handler that needs to be added to lambda must be part of the constant array ROUTES_TO_ADD within lambda_main.py file, otherwise handler will be ignored.

Note that `ROUTES_TO_ADD` assignment must follow this format: ROUTES_TO_ADD = [AdditionalHandler, ...]. Assignments from functions, other variables, etc. will cause validation error.

There are strict requirements for the format of custom handlers:

Handler must be declared within lambda_main.py
Handler must be inherited from LambdaUserHandler
Handler must have route attribute assignment within class body. Note that `route` must be assigned to a constant or formatted string, not to a function result or other variable
Handler must implement at least one of the async methods: post, get, put, patch, head, options, delete
If route includes path parameters, this parameters must be included into method signature with the same name. See routing format description here.
Custom route must not match default routes such as /healthcheck, /main, /docs/spec, /config

Note that `route` and http `post`, `get`, `put`, `patch`, `head`, `options`, `delete` methods must be declared within one final class (not in parent classes)

Breaking one of this requirement will cause lambda validation error

Some examples of handlers declaration.

route without parameters.

lambda_main.py

class AdditionalHandler(LambdaUserHandler):

    route: str = "/new_route"

    async def post(self):
        return self.sendResponse(201, json={"response": parameter})

ROUTES_TO_ADD = [AdditionalHandler]

route with two parameters.

lambda_main.py

class AdditionalHandler(LambdaUserHandler):

    route: str = r"/new_route/<parameterOne:.*>/<parameterTwo:.*>"

    async def post(self, parameterOne, parameterTwo):
        return self.sendResponse(201, json={"response": parameter})

ROUTES_TO_ADD = [AdditionalHandler]

Handler with inheritance.

lambda_main.py

class BaseAdditionalHandler(LambdaUserHandler):

    def getResponse(self, parameter):
        return {"response": parameter}

class AdditionalHandler(BaseAdditionalHandler):

    route: str = r"/new_route/<parameter:.*>"

    async def post(self, parameter):
        return self.sendResponse(201, json=self.getResponse(parameter))

ROUTES_TO_ADD = [AdditionalHandler]