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:
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()}
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:
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}
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.
from luna_lambda_tools import StandaloneLambdaRequest, logger
async def main(request: StandaloneLambdaRequest) -> dict:
logger.info("Hello, world!")
return {"result": "empty"}
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.
├──lambda_main.py
├──path
│ ├──fileinpath1.py
│ └──fileinpath2.py
└──file1.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(),
}
def foo_from_file1():
return "foo_from_file1"
from .fileinpath2 import variable
def foo_from_fileinpath1():
return "foo_from_fileinpath1"
def get_variable_from_another_file():
return variable
variable = 1
def foo_from_fileinpath2():
return "foo_from_fileinpath2"
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.
├──lambda_main.py
└──lambda_config.yml
You can import this configuration to your lambda code and use as python dictionary object.
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"
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
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:
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"])}
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:
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):
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
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)