Spaces:
Running
Running
File size: 1,845 Bytes
a0e37e2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
from typing import List, Dict, Union, Optional, Any
from time import sleep
import json
import boto3
class LambdaInvokeBase:
"""Base class for AWS Lambda direct-invocation based classes. Each class which inherits from this only serves a
single function.
Parameters
----------
function_name : str
Name of the Lambda function to invoke
access_key : Optional[str], optional
AWS access key, by default None
secret_key : Optional[str], optional
AWS secret key, by default None
"""
errors = frozenset([
"Unhandled"
])
def __init__(
self, function_name: str,
access_key: Optional[str] = None, secret_key: Optional[str] = None,
) -> None:
if access_key is not None and secret_key is not None:
self._client = boto3.client(
"lambda",
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
region_name="us-east-1",
)
else:
self._client = boto3.client("lambda", region_name='us-east-1')
self.function_name = function_name
def _submit_request(self, payload: Dict[str, Any]) -> Union[Dict[str, Any], List[Any]]:
response = self._client.invoke(
FunctionName=self.function_name,
InvocationType="RequestResponse",
Payload=json.dumps(payload),
)
if response.get("FunctionError") in self.errors:
# could use recursion, but we need to keep track of number of function calls
sleep(1)
response = self._client.invoke(
FunctionName=self.function_name,
InvocationType="RequestResponse",
Payload=json.dumps(payload),
)
return json.loads(response["Payload"].read())
|