Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use batch_execute_statement method in localstack

Best Python code snippet using localstack_python

dataapi_examples.py

Source:dataapi_examples.py

...125    except DataAccessLayerException as e:126        print(e)127# Introduce batch inserts128@timeit129def batch_execute_statement(sql, sql_parameter_sets):130    response = rds_client.batch_execute_statement(131        secretArn=db_credentials_secrets_store_arn,132        database=database_name,133        resourceArn=db_cluster_arn,134        sql=sql,135        parameterSets=sql_parameter_sets136    )137    return response138# Batch insert example139# Ask Data API what's the max batch size!140@timeit141def example_batch_insert():142    print('===== Example - Batch insert =====')143    sql = 'insert into package (package_name, package_version) values (:package_name, :package_version)'144    sql_parameter_sets = []145    for i in range(10,20):146        entry = [147                {'name':'package_name', 'value':{'stringValue': f'package{i}'}},148                {'name':'package_version', 'value':{'stringValue': 'version-1'}}149        ]150        sql_parameter_sets.append(entry)151    response = batch_execute_statement(sql, sql_parameter_sets)152    print(f'Number of records updated: {len(response["updateResults"])}')153# Transactions (commit and rollback)154# Here we redefine functions execute_statement() and batch_execute_statement() to support transactions155@timeit156def example_handling_transactions(package_start_idx, package_end_idx):157    @timeit158    def execute_statement(sql, sql_parameters=[], transaction_id=None):159        parameters = {160            'secretArn': db_credentials_secrets_store_arn,161            'database': database_name,162            'resourceArn': db_cluster_arn,163            'sql': sql,164            'parameters': sql_parameters165        }166        if transaction_id is not None:167            parameters['transactionId'] = transaction_id168        response = rds_client.execute_statement(**parameters)169        return response170    @timeit171    def batch_execute_statement(sql, sql_parameter_sets, transaction_id=None):172        parameters = {173            'secretArn': db_credentials_secrets_store_arn,174            'database': database_name,175            'resourceArn': db_cluster_arn,176            'sql': sql,177            'parameterSets': sql_parameter_sets178        }179        if transaction_id is not None:180            parameters['transactionId'] = transaction_id181        response = rds_client.batch_execute_statement(**parameters)182        return response183    print('===== Example - Handling transactions (commit and rollback) =====')184    transaction = rds_client.begin_transaction(185        secretArn=db_credentials_secrets_store_arn,186        resourceArn=db_cluster_arn,187        database=database_name)188    try:189        sql = 'insert into package (package_name, package_version) values (:package_name, :package_version)'190        sql_parameter_sets = []191        for i in range(package_start_idx,package_end_idx):192            entry = [193                    {'name':'package_name', 'value':{'stringValue': f'package-{i}'}},194                    {'name':'package_version', 'value':{'stringValue': 'version-1'}}195            ]196            sql_parameter_sets.append(entry)197        response = batch_execute_statement(sql, sql_parameter_sets, transaction['transactionId'])198    except Exception as e:199        print(f'Error: {e}')200        transaction_response = rds_client.rollback_transaction(201            secretArn=db_credentials_secrets_store_arn,202            resourceArn=db_cluster_arn,203            transactionId=transaction['transactionId'])204    else:205        transaction_response = rds_client.commit_transaction(206            secretArn=db_credentials_secrets_store_arn,207            resourceArn=db_cluster_arn,208            transactionId=transaction['transactionId'])209        print(f'Number of records updated: {len(response["updateResults"])}')210    print(f'Transaction Status: {transaction_response["transactionStatus"]}')211# Running our examples in order...

Copy_s3_to_rds_arora_Serverless.py

Source:Copy_s3_to_rds_arora_Serverless.py

...14            sql=sql,15            parameters=sql_parameters16        )17        return response18    def batch_execute_statement(sql, sql_parameter_sets):19        response = rds_client.batch_execute_statement(20            secretArn=db_credentials_secrets_store_arn,21            database=database_name,22            resourceArn=db_cluster_arn,23            sql=sql,24            parameterSets=sql_parameter_sets25        )26        return response27    '''28    ## Parameterised Insert29    sql = 'insert into package (package_name, package_version) values (:package_name, :package_version)'30    sql_parameters = [31        {'name':'package_name', 'value':{'stringValue': 'package-2'}},32        {'name':'package_version', 'value':{'stringValue': 'version-1'}}33    ]34    response = execute_statement(sql, sql_parameters)35    print(f'Number of records updated: {response["numberOfRecordsUpdated"]}')36    '''37    ## Simple Get all querry38    response = execute_statement('select * from package')39    print(response['records'])40    '''41    ## Parameterised Get42    sql = 'select * from package where package_name=:package_name'43    package_name = 'package2'44    sql_parameters = [{'name':'package_name', 'value':{'stringValue': f'{package_name}'}}]45    response = execute_statement(sql, sql_parameters)46    print(response['records'])47    ## Batch Insert48    sql = 'insert into package (package_name, package_version) values (:package_name, :package_version)'49    sql_parameter_sets = []50    for i in range(1,11):51        entry = [52                    {'name':'package_name', 'value':{'stringValue': f'package{i}'}},53                    {'name':'package_version', 'value':{'stringValue': f'version{i}'}}54            ]55        sql_parameter_sets.append(entry)56    response = batch_execute_statement(sql, sql_parameter_sets)57    print(f'Number of records updated: {len(response["updateResults"])}')58    ## Committing or roll back.59    transaction = rds_client.begin_transaction(60        secretArn=db_credentials_secrets_store_arn,61        resourceArn=db_cluster_arn,62        database=database_name)63    try:64        sql = 'insert into package (package_name, package_version) values (:package_name, :package_version)'65        sql_parameter_sets = []66        for i in range(30,40):67            entry = [68                    {'name':'package_name', 'value':{'stringValue': f'package{i}'}},69                    {'name':'package_version', 'value':{'stringValue': f'version{i}'}}70            ]71            sql_parameter_sets.append(entry)72        response = batch_execute_statement(sql, sql_parameter_sets, transaction['transactionId'])73    except Exception:74        transaction_response = rds_client.rollback_transaction(75            secretArn=db_credentials_secrets_store_arn,76            resourceArn=db_cluster_arn,77            transactionId=transaction['transactionId'])78    else:79        transaction_response = rds_client.commit_transaction(80            secretArn=db_credentials_secrets_store_arn,81            resourceArn=db_cluster_arn,82            transactionId=transaction['transactionId'])83        print(f'Number of records updated: {len(response["updateResults"])}')84    print(f'Transaction Status: {transaction_response["transactionStatus"]}')...

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.