(self, function_arn: str)
| 508 | return False |
| 509 | |
| 510 | def status(self, function_arn: str) -> Optional[Dict[str, Any]]: |
| 511 | try: |
| 512 | response = self._s3.get_bucket_notification_configuration(Bucket=self.bucket_name) |
| 513 | lambda_configs = response.get("LambdaFunctionConfigurations", []) |
| 514 | notification_id = self._make_notification_id(function_arn) |
| 515 | for config in lambda_configs: |
| 516 | if config.get("Id") == notification_id: |
| 517 | return config |
| 518 | return None |
| 519 | except Exception: |
| 520 | LOG.exception("Unable to get S3 event source status") |
| 521 | return None |
| 522 | |
| 523 | def update(self, function_arn: str) -> None: |
| 524 | # For S3, update is remove + add |
no test coverage detected