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

How to use kms_key method in localstack

Best Python code snippet using localstack_python

outputs.py

Source:outputs.py

...33        """34        return pulumi.get(self, "encryption_type")35    @property36    @pulumi.getter(name="kmsKey")37    def kms_key(self) -> Optional[str]:38        """39        The ARN of the KMS key to use when `encryption_type` is `KMS`. If not specified, uses the default AWS managed key for ECR.40        """41        return pulumi.get(self, "kms_key")42    def _translate_property(self, prop):43        return _tables.CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop44@pulumi.output_type45class RepositoryImageScanningConfiguration(dict):46    def __init__(__self__, *,47                 scan_on_push: bool):48        """49        :param bool scan_on_push: Indicates whether images are scanned after being pushed to the repository (true) or not scanned (false).50        """51        pulumi.set(__self__, "scan_on_push", scan_on_push)52    @property53    @pulumi.getter(name="scanOnPush")54    def scan_on_push(self) -> bool:55        """56        Indicates whether images are scanned after being pushed to the repository (true) or not scanned (false).57        """58        return pulumi.get(self, "scan_on_push")59    def _translate_property(self, prop):60        return _tables.CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop61@pulumi.output_type62class GetRepositoryEncryptionConfigurationResult(dict):63    def __init__(__self__, *,64                 encryption_type: str,65                 kms_key: str):66        """67        :param str encryption_type: The encryption type to use for the repository, either `AES256` or `KMS`.68        :param str kms_key: If `encryption_type` is `KMS`, the ARN of the KMS key used.69        """70        pulumi.set(__self__, "encryption_type", encryption_type)71        pulumi.set(__self__, "kms_key", kms_key)72    @property73    @pulumi.getter(name="encryptionType")74    def encryption_type(self) -> str:75        """76        The encryption type to use for the repository, either `AES256` or `KMS`.77        """78        return pulumi.get(self, "encryption_type")79    @property80    @pulumi.getter(name="kmsKey")81    def kms_key(self) -> str:82        """83        If `encryption_type` is `KMS`, the ARN of the KMS key used.84        """85        return pulumi.get(self, "kms_key")86@pulumi.output_type87class GetRepositoryImageScanningConfigurationResult(dict):88    def __init__(__self__, *,89                 scan_on_push: bool):90        """91        :param bool scan_on_push: Indicates whether images are scanned after being pushed to the repository.92        """93        pulumi.set(__self__, "scan_on_push", scan_on_push)94    @property95    @pulumi.getter(name="scanOnPush")...

file_splitter_handler.py

Source:file_splitter_handler.py

1import os2from datetime import datetime3import logging4from logging.config import fileConfig5from lib.logger import get_logger6from os.path import join7import sys8import csv9import xlsxwriter as xw10import boto311from lib.settings import ENVIRONMENT, LOG_LEVEL,log_group_name_consumer,request_payload_region,ecr_repository_name,target_bucket,kms_key12fileConfig("resources/logging_config.ini")13logging.getLogger().setLevel(LOG_LEVEL)14logging.info("Preparing parameters")15## For Testing purpose 16bucket_name='anl-vip-dev-s3-input-filesplitter-src'17#file_name='input-file-1.csv'18#file_save_as=file_name.split('.')[0]+'_tmp'+'.'+file_name.split('.')[1]19#kms_key='7912073d-d5ce-4dc7-8034-4ddad9e29ae3'20#tgt_bucket_name = 'anl-vip-dev-s3-input-filesplitter-tgt'21######################22parameters = {23    "ENVIRONMENT": ENVIRONMENT,24    "log_group_name_consumer": log_group_name_consumer,25    "request_payload_region" : request_payload_region,26    "ecr_repository_name" : ecr_repository_name,27    #"source_bucket_name" : source_bucket, 28    "tgt_bucket_name" : target_bucket,29    "kms_key": kms_key30}31#bucket_name = parameters["source_bucket_name"]32tgt_bucket_name = parameters["tgt_bucket_name"]33file_name='input-file-1.csv'34file_save_as=file_name.split('.')[0]+'_tmp'+'.'+file_name.split('.')[1]35kms_key=parameters["kms_key"]36s3_client = boto3.client('s3',region_name='eu-west-1')37def get_dict_list(bucket_name,file_name,file_save_as):38    #s3_client.get_object(Bucket='anl-vip-dev-s3-input-filesplitter-src',Key='input-file-1.csv')39    s3_client.download_file(bucket_name,file_name,file_save_as)40    logging.info('File downloaded successfully ....')41    logging.info(os.listdir('.'))42    #s3_client.put_object()43    list_of_dict_items = []44    with open(file_save_as, encoding="utf8") as f:45        csv_reader = csv.DictReader(f,delimiter=";")46        for line in csv_reader:47            result_dict = dict(line)48            list_of_dict_items.append(result_dict)49        50        return list_of_dict_items51        52def write_to_excel_output(input_dict):53    today_date = datetime.today().strftime('%Y%m%d')54    get_id = input_dict.get('contract_nr')55    output_file_name= str(get_id)+'_'+str(today_date)+'.xlsx'56    logging.info(f'File : {output_file_name}')57    workbook = xw.Workbook(output_file_name)58    worksheet = workbook.add_worksheet()59    col_num = 060    for key, value in input_dict.items():61        worksheet.write(0, col_num, key)62        worksheet.write(1, col_num, value)63        col_num += 164    workbook.close()65def write_files_to_s3(bucket_name,kms_key):66    67    files = [f for f in os.listdir('.') if os.path.isfile(f)]68    for f in files:69        #logging.info(f)70        file_extension = f.split('.')[1]71        if file_extension == 'xlsx':72            object_key = 'tmp/'+ f73            s3_client.upload_file(f,bucket_name,object_key,ExtraArgs={"ServerSideEncryption": "aws:kms", "SSEKMSKeyId": kms_key })74            logging.info(f'File {f} is written successfully ...')75if __name__ == "__main__":76    logging.info("-------------------------- FILE SPLITTER PROCESS BEGINS HERE  -------------------------- ")77    get_list_of_dict = get_dict_list(bucket_name,file_name,file_save_as)78    for item in get_list_of_dict:79        write_to_excel_output(item)80    81    write_files_to_s3(tgt_bucket_name,kms_key)...

_inputs.py

Source:_inputs.py

...34    def encryption_type(self, value: Optional[pulumi.Input[str]]):35        pulumi.set(self, "encryption_type", value)36    @property37    @pulumi.getter(name="kmsKey")38    def kms_key(self) -> Optional[pulumi.Input[str]]:39        """40        The ARN of the KMS key to use when `encryption_type` is `KMS`. If not specified, uses the default AWS managed key for ECR.41        """42        return pulumi.get(self, "kms_key")43    @kms_key.setter44    def kms_key(self, value: Optional[pulumi.Input[str]]):45        pulumi.set(self, "kms_key", value)46@pulumi.input_type47class RepositoryImageScanningConfigurationArgs:48    def __init__(__self__, *,49                 scan_on_push: pulumi.Input[bool]):50        """51        :param pulumi.Input[bool] scan_on_push: Indicates whether images are scanned after being pushed to the repository (true) or not scanned (false).52        """53        pulumi.set(__self__, "scan_on_push", scan_on_push)54    @property55    @pulumi.getter(name="scanOnPush")56    def scan_on_push(self) -> pulumi.Input[bool]:57        """58        Indicates whether images are scanned after being pushed to the repository (true) or not scanned (false)....

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.