yalambda
Yalambda lets you write Yanex.Cloud Functions with less boilerplate
Features:
- everything is type-annotated, so you'll get autocompletion in IDEs
- base64 de/encoding and other details are handled for you
- automatically parse JSON using
dataclass-factory
Echo server example
from yalambda import function, YaRequest, YaResponse
@function()
async def handler(req: YaRequest) -> YaResponse:
return YaResponse(200, req.body)Automatically parse dataclasses
from dataclasses import dataclass
from yalambda import function, YaResponse
@dataclass
class Point:
x: float
y: float
@function()
async def handler(point: Point) -> YaResponse:
dist = (point.x**2 + point.y**2)**0.5
return YaResponse(200, {"distance_to_origin": dist})Access both the dataclass and the request
from dataclasses import dataclass
from yalambda import function, YaRequest, YaResponse
@dataclass
class Point:
x: float
y: float
@function()
async def handler(point: Point, req: YaRequest) -> YaResponse:
if req.http_method != "POST":
return YaResponse(405, "Only POST requests are allowed")
dist = (point.x**2 + point.y**2)**0.5
return YaResponse(200, {"distance_to_origin": dist})Initialize something asynchronously on first call
from yalambda import function, YaRequest, YaResponse
async def init():
global answer
answer = 42
@function(init)
async def handler(req: YaRequest) -> YaResponse:
return YaResponse(200, "Answer: {}".format(answer))Routing
from dataclasses import dataclass
from yalambda import dispatch, YaRequest, YaResponse
@dataclass
class Point:
x: float
y: float
async def get_all_points(req: YaRequest) -> YaResponse:
points = [{"x": 3.0, "y": 4.0}, {"x": -1.0, "y": 3.27}]
return YaResponse(200, points)
async def compute_distance(point: Point) -> YaResponse:
dist = (point.x**2 + point.y**2)**0.5
return YaResponse(200, {"distance_to_origin": dist})
handler = dispatch({
"GET": get_all_points,
"POST": compute_distance,
})Full example
This function acts as a GitHub webhook and sends a pretty embed on Discord webhook when an issue is opened or closed. See the source code on GitHub.
Development server
You can install aiohttp and run your function locally.
It's not the same as the real thing, but it should be enough for simple functions.
$ python -m yalambda your_module
======== Running on http://0.0.0.0:55710 ========
(Press CTRL+C to quit)Condition DSL
We can modify our GitHub->Discord example so that it doesn't error out on the initial ping event:
from yalambda import when
...
async def handle_issue_events(event: IssueEvent) -> YaResponse:
embed = create_embed(event)
if embed is not None:
await client.post(DISCORD_WEBHOOK, json={"embeds": [embed]})
return YaResponse(200, "")
async def handle_ping(req: YaRequest) -> YaResponse:
return YaResponse(200, "")
handler = when.dispatch(
when.header_is("x-github-event", "ping", handle_ping),
when.header_is("x-github-event", "issues", handle_issue_events),
init=init
)
