File size: 2,086 Bytes
b404de3 | 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 60 61 62 63 64 65 66 67 68 | from __future__ import annotations
from typing import Any, Dict, List
from cloudenv.server.providers.aws.base import BaseService
class LambdaService(BaseService):
_service_name = "lambda"
_boto_service = "lambda"
def _infer_instance_ids(self, payload: Dict[str, Any]) -> List[str]:
fn = payload.get("FunctionName")
return [fn] if fn else []
def _execute(
self, action: str, payload: Dict[str, Any], instance_ids: List[str]
) -> Any:
if action == "AddPermission":
fn = payload.get("FunctionName")
if fn:
try:
self.enforcer.attach_instance_policy_document(
"lambda",
fn,
{
"Version": "2012-10-17",
"Statement": [payload],
},
)
except Exception:
pass
elif action in ("RemovePermission", "DeleteFunction"):
fn = payload.get("FunctionName")
if fn:
try:
self.enforcer.set_instance_policies("lambda", fn, [])
except Exception:
pass
return super()._execute(action, payload, instance_ids)
def reset(self) -> None:
try:
funcs = self.client.list_functions().get("Functions", [])
for f in funcs:
try:
self.client.delete_function(FunctionName=f["FunctionName"])
except Exception:
pass
except Exception:
pass
def health(self) -> float:
try:
self.client.list_functions()
return 1.0
except Exception:
return 0.0
def describe_state(self) -> Dict[str, Any]:
try:
funcs = self.client.list_functions().get("Functions", [])
return {"functions": [f["FunctionName"] for f in funcs]}
except Exception:
return {"functions": []}
|