This is a very simple package which adds a few decorators to simplify the use of a single lambda function for multiple
endpoints for AWS Api Gateway. It uses the operationId
attribute in the OAS3 specification to identify the correct
handler and then passes the event and context to the correct method.
import json
from sfunc import SFunc
sfunc = SFunc()
@sfunc.operation('getPetsByName')
def get_pets_by_name(event, context):
"""The methods need to have the same signature as a regular lambda handler entrypoint"""
return {
'statusCode': 200,
'body': json.dumps({'names': ['fido', 'catty']})
}
@sfunc.operation('getPetsByAge')
def get_pets_by_age(event, context):
"""This will be invoked when the endpoint with the operationId 'getPetsByAge' is invoked"""
return {
'statusCode': 204
}
def lambda_handler(event, context):
"""We only need to specify one entry point and can ignore any other """
rsp = sfunc.execute(event, context) # rsp will contain the return value from the invoked method
return rsp
For request/response which doesn't require any additional and are json encoded, sfunc provides two convenience classes for request and response which can simplify the code further when it comes to parsing and getting headers and path params. The following code will result in the same outcome as the example above, except for the extra headers.
from sfunc import ApiGatewayRequest, ApiGatewayResponse, SFunc
sfunc = SFunc()
@sfunc.operation('getPetsByName')
def pets_by_name(req: ApiGatewayRequest, context) -> ApiGatewayResponse:
_ = req.headers().get('x-request-id')
return ApiGatewayResponse(200, {'names': ['fido', 'catty']})
@sfunc.operation('getPetsByAge')
def pets_by_age(req: ApiGatewayRequest, context) -> ApiGatewayResponse:
rsp = ApiGatewayResponse(204)
rsp.add_header('content-type', 'application/json')
return rsp
def lambda_handler(event, context):
return sfunc.sfunc_execute(ApiGatewayRequest(event), context).response()
It might be useful during development or if some endpoints are missing the operationId attribute in the specification.
from sfunc import SFunc
sfunc = SFunc()
@sfunc.default()
def default_handler(event, context):
return {
'statusCode': 405
}
def lambda_handler(event, context):
return sfunc.execute(event, context) # or sfunc.sfunc_execute(ApiGatewayRequest) -> ApiGatewayResponse as above