from __future__ import annotations import json from typing import TYPE_CHECKING, Any, Literal import boto3 from fansifter_common.adapters.aws.secretsmanager import SecretsManager from fansifter_common.utils.functional import lazy_proxy from app.config import settings if TYPE_CHECKING: from mypy_boto3_lambda import LambdaClient as _LambdaClient class LambdaClient: def __init__( self, *, region_name: str, session: boto3.Session | None = None ) -> None: self.region_name = region_name self._session = session self._client: _LambdaClient | None = None @property def session(self) -> boto3.Session: if self._session is None: self._session = boto3.Session() return self._session @property def client(self) -> _LambdaClient: if self._client is None: self._client = self.session.client("lambda", region_name=self.region_name) return self._client def invoke( self, *, function_name: str, data: dict[str, Any], invocation_type: Literal["DryRun", "Event", "RequestResponse"], ) -> bool: response = self.client.invoke( FunctionName=function_name, InvocationType=invocation_type, Payload=json.dumps(data), ) return 200 <= response["StatusCode"] < 300 def get_lambda_client() -> LambdaClient: return LambdaClient(region_name=settings.aws_region_name) lambda_client = lazy_proxy(get_lambda_client) def get_secrets_manager() -> SecretsManager: return SecretsManager(region_name=settings.aws_region_name) secrets_manager = lazy_proxy(get_secrets_manager)