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

How to use update_global_table method in localstack

Best Python code snippet using localstack_python

cfn-global-table.py

Source:cfn-global-table.py

...90        AddReplicationGroups = new_rep_groups - old_rep_groups91        RemoveReplicationGroups = old_rep_groups - new_rep_groups92        if len(AddReplicationGroups):93            for region in AddReplicationGroups:94                update_global_table(global_table_name, region, 'Create')95                logger.info('Add replication group region: ' + region)96        if len(RemoveReplicationGroups):97            for region in RemoveReplicationGroups:98                update_global_table(global_table_name, region, 'Delete')99                logger.info('Remove replication group region: ' + region)100        resp['Reason'] = 'Updated Global Table: ' + global_table_name101        resp['Status'] = 'SUCCESS'102    except Exception as e:103        logger.error('Failed to update DynamoDB Global Table: ' + global_table_name)104        resp['Reason'] = 'Failed to update Global Table: ' + e.message105        resp['Status'] = 'FAILED'106    107    return resp108def update_global_table(global_table_name, region, action):109    client = boto3.client('dynamodb')110    ReplicaUpdates = []111    ReplicaUpdate = {}112    ReplicationGroup = {}113    ReplicationGroup['RegionName'] = region114    ReplicaUpdate[action] = ReplicationGroup115    ReplicaUpdates.append(ReplicaUpdate)116    response = client.update_global_table(117        GlobalTableName=global_table_name,118        ReplicaUpdates=ReplicaUpdates119    )120    return response121def handler(event, context):122  if event['RequestType'] == 'Create':123    response = global_table_create(event,context)124  if event['RequestType'] == 'Delete':125    response = {'Status': 'SUCCESS', 'Data': { 'LogGroup': context.log_group_name }}126  if event['RequestType'] == 'Update':127    response = global_table_update(event,context)128  if 'Reason' in response.keys():129    cfnresponse.send(event, context, log_level, response['Status'], response['Data'], response['Reason'])130  else:...

dynamoDBGlobalTableCreate.py

Source:dynamoDBGlobalTableCreate.py

...87        AddReplicationGroups = new_rep_groups - old_rep_groups88        RemoveReplicationGroups = old_rep_groups - new_rep_groups89        if len(AddReplicationGroups):90            for region in AddReplicationGroups:91                update_global_table(global_table_name, region, 'Create')92                logger.info('Add replication group region: ' + region)93        if len(RemoveReplicationGroups):94            for region in RemoveReplicationGroups:95                update_global_table(global_table_name, region, 'Delete')96                logger.info('Remove replication group region: ' + region)97        resp['Reason'] = 'Updated Global Table: ' + global_table_name98        resp['Status'] = 'SUCCESS'99    except Exception as e:100        logger.error('Failed to update DynamoDB Global Table: ' + global_table_name)101        resp['Reason'] = 'Failed to update Global Table: ' + e.message102        resp['Status'] = 'FAILED'103    104    return resp105def update_global_table(global_table_name, region, action):106    client = boto3.client('dynamodb')107    ReplicaUpdates = []108    ReplicaUpdate = {}109    ReplicationGroup = {}110    ReplicationGroup['RegionName'] = region111    ReplicaUpdate[action] = ReplicationGroup112    ReplicaUpdates.append(ReplicaUpdate)113    response = client.update_global_table(114        GlobalTableName=global_table_name,115        ReplicaUpdates=ReplicaUpdates116    )117    return response118def handler(event, context):119  if event['RequestType'] == 'Create':120    response = global_table_create(event,context)121  if event['RequestType'] == 'Delete':122    response = {'Status': 'SUCCESS', 'Data': { 'LogGroup': context.log_group_name }}123  if event['RequestType'] == 'Update':124    response = global_table_update(event,context)125  if 'Reason' in response.keys():126    cfnresponse.send(event, context, log_level, response['Status'], response['Data'], response['Reason'])127  else:...

index.py

Source:index.py

...33            print("done")34        except boto_client.exceptions.GlobalTableAlreadyExistsException:35            try:36                print("Global table already exists, trying to join this region to the global table")37                global_table = boto_client.update_global_table(38                    GlobalTableName=self.table_name,39                    ReplicaUpdates=[40                        {'Create': {'RegionName': REGION}}41                    ],42                )43                print("done")44            except boto_client.exceptions.ReplicaAlreadyExistsException:45                raise RuntimeError("This region is already joined to the GlobalTable. "46                                   "Not taking ownership of this Join.")47        self.physical_resource_id = global_table['GlobalTableDescription']['GlobalTableArn']48        print(f"GlobalTableArn: {self.physical_resource_id}")49        return {}50    def update(self):51        if self.has_property_changed('TableName'):52            # We need a new GlobalTable, switch to create and let CLEANUP delete the old one53            return self.create()54        # Nothing else can change55        # Ignore request succesfully56        print("Ignoring update")57        return {}58    def delete(self):59        boto_client = self.get_boto3_client('dynamodb')60        try:61            print("Trying to delete global table {self.table_name}")62            boto_client.update_global_table(63                GlobalTableName=self.table_name,64                ReplicaUpdates=[65                    {'Delete': {'RegionName': REGION}}66                ],67            )68            print("done")69        except boto_client.exceptions.GlobalTableNotFoundException:70            print("Delete failed, assuming success...")...

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.