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.

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:

lambda_main.py
from luna_lambda_tools import StandaloneLambdaRequest


async def main(request: StandaloneLambdaRequest) -> dict:
    versions = {
        "faces_address": request.clients.faces.getAddress(),
        "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)