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