| 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": []} | |