KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use put_bucket_notification method in localstack

Best Python code snippet using localstack_python

lambda_handler.py

Source:lambda_handler.py

...33    logger = Logger()34    s3 = boto3.client("s3")35except Exception as e:36    helper.init_failure(e)37def put_bucket_notification(bucket: str, config: Dict[str, Any]) -> None:38    """39    Put an S3 bucket notification configuration40    """41    try:42        s3.put_bucket_notification_configuration(43            Bucket=bucket, NotificationConfiguration=config44        )45        logger.debug(f"Successfully put bucket notification to s3://{bucket}")46    except botocore.exceptions.ClientError:47        logger.exception(f"Unable to put bucket notification to s3://{bucket}")48        raise49@helper.create50@helper.update51def create(event: Dict[str, Any], context: LambdaContext) -> Optional[str]:52    props = event.get("ResourceProperties", {})53    bucket_name = props.get("BucketName")54    if not bucket_name:55        raise ValueError("BucketName has not been provided")56    config = props.get("NotificationConfiguration", {})57    if not config:58        raise ValueError("NotificationConfiguration has not been provided")59    put_bucket_notification(bucket=bucket_name, config=config)60    return "ResultsNotifications"61@helper.delete62def delete(event: Dict[str, Any], context: LambdaContext) -> None:63    props = event.get("ResourceProperties", {})64    bucket_name = props.get("BucketName")65    if not bucket_name:66        raise ValueError("BucketName has not been provided")67    # empty dict removes bucket notification68    put_bucket_notification(bucket=bucket_name, config={})69@logger.inject_lambda_context(log_event=True)70def handler(event: Dict[str, Any], context: LambdaContext) -> None:...

put_bucket_notification.py

Source:put_bucket_notification.py

1"""2Boto3 python script to add permission for s3 to invoke lambda and also send a notification to lambda function when a csv file is uploaded to s3 bucket3Required 3rd party python packages that can be installed with pip: boto3, docopt4python3 -m pip install <package-name>5Usage: 6  put_bucket_notification.py <BucketName> -l lambda_fn_name -a aws_acc_id -r aws_region7  put_bucket_notification.py -h | --help8Options:9  -l lambda_fn_name10  -a aws_acc_id11  -r aws_region12  -h --help13"""14import boto315from docopt import docopt16bucket_name = None17lambda_arn = None18if __name__ == '__main__':19    arguments = docopt(__doc__)20    if arguments["<BucketName>"] != None:21        bucket_name = arguments["<BucketName>"]22    if arguments["-l"] != None:23        lambda_function_name = arguments["-l"]24    if arguments["-a"] != None:25        aws_account_id = arguments["-a"]26    if arguments["-r"] != None:27        region = arguments["-r"]28lambda_client = boto3.client('lambda')29try:30    permit_s3_lambda = lambda_client.add_permission(31        Action='lambda:InvokeFunction',32        FunctionName=lambda_function_name,33        Principal='s3.amazonaws.com',34        SourceAccount=aws_account_id,35        SourceArn='arn:aws:s3:::'+bucket_name,36        StatementId='s3',37    )38except Exception as e:39    print("S3 bucket already has permission to invoke lambda")40aws_s3_client = boto3.client('s3')41response = aws_s3_client.put_bucket_notification_configuration(42    Bucket=bucket_name,43    NotificationConfiguration={44        45        'LambdaFunctionConfigurations': [46            {47               #Optional 'Id': 'string',48                'LambdaFunctionArn': "arn:aws:lambda:"+region+":"+aws_account_id+":function:"+lambda_function_name, #Required49                'Events': [50                    's3:ObjectCreated:*',51                ],52                'Filter': {53                    'Key': {54                        'FilterRules': [55                            {56                                'Name': 'suffix',57                                'Value': '.csv'58                            },59                        ]60                    }61                }62            },63        ]64    }...

main.py

Source:main.py

...20    )21notify = {22            23}24s3_client.put_bucket_notification(25        Bucket="my-bucket-aris", NotificationConfiguration=notify26    )27s3_client.put_public_access_block(28    Bucket='my-bucket-aris',29    PublicAccessBlockConfiguration={30        'BlockPublicAcls': True,31    }32)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.