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

How to use create_global_table method in localstack

Best Python code snippet using localstack_python

db.py

Source:db.py

...11        )12        self.cursor = self.db.cursor()13    def init_db(self, db_name: str):14        if not self.table_exists("global"):15            self.create_global_table()16        if not self.table_exists("config"):17            self.create_config_table()18        # self.cursor.execute("SHOW DATABASES")19        # dbs = self.cursor.fetchall()20        # found = False21        # for item in dbs:22        #     if item[0].lower() == db_name.lower():23        #         found = True24        #         break25        # if found:26        #     print("Database Found")27        #     self.cursor.execute(f"USE {db_name}")28        # else:29        #     print("Creating Database...")30        #     self.create_db(db_name)31    def create_db(self, db_name: str) -> None:32        self.cursor.execute(f"CREATE DATABASE {db_name}")33        self.cursor.execute(f"USE {db_name}")34        self.db.commit()35        self.create_global_table()36        # Create Config Table as well37        self.create_config_table()38        print("Database Created!")39    def create_global_table(self) -> None:40        self.db.ping(reconnect=True, attempts=5)41        self.cursor.execute(42            f"CREATE TABLE global(leaderboard VARCHAR(100) NOT NULL, symbol VARCHAR(50) DEFAULT ':coin:')")43        self.db.commit()44    def create_config_table(self) -> None:45        self.db.ping(reconnect=True, attempts=5)46        self.cursor.execute(47            "CREATE TABLE config(property VARCHAR(50), value VARCHAR(70))")48        self.cursor.executemany(49            "INSERT INTO config VALUES(%s, %s)", [50                ["user_role", "bot-user"],51                ["color_success", "0x0084ff"],52                ["color_error", "0x991b1b"],53                ["broadcast_channel", ""]54            ]55        )56        self.db.commit()57    def create_table(self, name: str, symbol: str = ":coin:") -> dict:58        name = name.lower()59        self.db.ping(reconnect=True, attempts=5)60        if self.table_exists(name):61            return {62                "success": False,63                "message": "Leaderboard Already Exists"64            }65        self.cursor.execute(66            f"CREATE TABLE {name}(user_id BIGINT PRIMARY KEY, user_name VARCHAR(50) NOT NULL, points INT DEFAULT 0) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci")67        self.cursor.execute(68            f"INSERT INTO global VALUES('{name}', '{symbol}')"69        )70        self.db.commit()71        print(f"Created Table {name}")72        return {73            "success": True,74            "message": f"Created Leaderboard: {name}"75        }76    def add_user_to_table(self, leaderboard: str, user_id: int, user_name: str) -> dict:77        leaderboard = leaderboard.lower()78        self.db.ping(reconnect=True, attempts=5)79        self.cursor.execute(80            f"SELECT * FROM {leaderboard} WHERE user_id = '{user_id}'")81        user = self.cursor.fetchone()82        if user:83            print("User Already Exists!")84            return {85                "success": False,86                "message": f"User {user_name} Already exists in {leaderboard}"87            }88        self.cursor.execute(89            f"INSERT INTO {leaderboard}(user_id, user_name) VALUES('{user_id}', '{user_name}')")90        print("Added User")91        self.db.commit()92        return {93            "success": True,94            "message": f"Created User {user_name} in {leaderboard}"95        }96    def add_points_to_user(self, leaderboard: str, user_id: int, amount: int, user_name: str) -> dict:97        leaderboard = leaderboard.lower()98        self.db.ping(reconnect=True, attempts=5)99        self.cursor.execute(100            f"SELECT * FROM {leaderboard} WHERE user_id = '{user_id}'")101        user = self.cursor.fetchone()102        if not user:103            print(f"User not found in {leaderboard}")104            self.add_user_to_table(leaderboard, user_id, user_name)105        self.cursor.execute(106            f"UPDATE {leaderboard} SET points = points + {amount} WHERE user_id = '{user_id}'")107        self.db.commit()108        symbol = self.get_symbol(leaderboard)109        print(f"Added {amount} {symbol} to {user_id}")110        return {111            "success": True,112            "message": None,113            "symbol": symbol114        }115    def get_leaderboard(self, leaderboard: str, page: int = 1, max: int = 25) -> dict:116        leaderboard = leaderboard.lower()117        self.db.ping(reconnect=True, attempts=5)118        self.cursor.execute(119            f"SELECT * FROM {leaderboard} ORDER BY points DESC LIMIT {max} OFFSET {(page - 1) * max}")120        data = self.cursor.fetchall()121        symbol = self.get_symbol(leaderboard)122        if len(data) == 0:123            return {124                "success": False,125                "message": "Page does not exist",126                "symbol": symbol127            }128        return {129            "success": True,130            "message": data,131            "symbol": symbol132        }133    def get_leaderboards(self) -> dict:134        self.db.ping(reconnect=True, attempts=5)135        self.cursor.execute("SHOW TABLES")136        tables = self.cursor.fetchall()137        return {138            "success": True,139            "message": tables140        }141    def delete_leaderboard(self, leaderboard: str) -> dict:142        leaderboard = leaderboard.lower()143        self.db.ping(reconnect=True, attempts=5)144        if not self.table_exists(leaderboard):145            return {146                "success": False,147                "message": f"Leaderboard {leaderboard} not found!"148            }149        self.cursor.execute(f"DROP TABLE {leaderboard}")150        self.cursor.execute(151            f"DELETE FROM global WHERE leaderboard LIKE '{leaderboard}'")152        self.db.commit()153        return {154            "success": True,155            "message": f"Deleted Leaderboard {leaderboard}"156        }157    def clear_db(self) -> dict:158        """Deletes all the tables from the database"""159        self.db.ping(reconnect=True, attempts=5)160        self.cursor.execute("SHOW TABLES")161        tables = self.cursor.fetchall()162        for table in tables:163            table_name = table[0]164            if table_name == "config":165                continue166            self.cursor.execute(f"DROP TABLE {table_name}")167        self.db.commit()168        self.create_global_table()169        return {170            "success": True,171            "message": "Deleted all the leaderboards"172        }173    def get_config(self, option: str = "") -> dict:174        self.db.ping(reconnect=True, attempts=5)175        query = "SELECT * FROM config"176        if option != "":177            query += f" WHERE property LIKE '{option}'"178        self.cursor.execute(query)179        values = self.cursor.fetchall()180        if len(values) == 0:181            return {182                "success": False,...

SemanticProcessor BACKUP.py

Source:SemanticProcessor BACKUP.py

1from Symbol_Table import Symbol_Table, Entry2import copy3class SemanticProcessor(object):4    def __init__(self):5        # symbol tables,6        self.level = 07        self.SymbolTable_stack = []8        # hold attrs while processing a table entry using the grammar rules9        self.prevToken_buffer = ''10        self.fParam_buffer = []11        self.entry_buffer = Entry(self.level, '', '', '')12        # the semantic functions dictionary to use when a semantic symbol is poped13        self.dispatcher = {'CREATE_GLOBAL_TABLE': self.CREATE_GLOBAL_TABLE,14                           'CLASS_ENTRY_TABLE': self.CLASS_ENTRY_TABLE,15                           'END_CLASS': self.END_CLASS,16                           'ENTRY_TYPE': self.ENTRY_TYPE,17                           'ENTRY_NAME': self.ENTRY_NAME,18                           'ADD_DECL_ARRAY_DIM': self.ADD_DECL_ARRAY_DIM,19                           'CLASS_VAR_ENTRY': self.CLASS_VAR_ENTRY,20                           'FUNC_ENTRY_TABLE': self.FUNC_ENTRY_TABLE,21                           # 'PARAM_TYPE': self.PARAM_TYPE,22                           # 'PARAM_NAME': self.PARAM_NAME,23                           # 'PARAM_NAME': self.PARAM_NAME,24                           'ADD_FUNC_PARAM_ENTRY': self.ADD_FUNC_PARAM_ENTRY,25                           'END_CLASS_FUNC': self.END_CLASS_FUNC,26                           'PROGRAM_FUNC_ENTRY_TABLE': self.PROGRAM_FUNC_ENTRY_TABLE27                           }28    def __str__(self):29        """String representation of the class instance."""30        return 'test'31    def __repr__(self):32        return self.__str__()33    def clearEntryBuffer(self):34        self.entry_buffer.name = ''35        self.entry_buffer.type = ''36        self.entry_buffer.kind = ''37        self.entry_buffer.arraySize = []38        self.entry_buffer.link = None39    ''''''''''''''''''''''''''''''40    '''   SEMANTIC FUNCTIONS   '''41    ''''''''''''''''''''''''''''''42    def CREATE_GLOBAL_TABLE(self):43        self.SymbolTable_stack.append(Symbol_Table(self.level, 'global'))44    ##################################################################45    #                     CLASS SEMANTIC ACTIONS                     #46    ##################################################################47    def CLASS_ENTRY_TABLE(self):48        print "create_classEntryAndTable."49        # create a table entry and link it to the new class table50        class_entry = Entry(self.level, self.prevToken_buffer, 'class', '')51        self.level += 152        class_entry.link = Symbol_Table(self.level, class_entry.name)53        # append the new entry to the global table and put the reference to the class table on top the stack54        self.SymbolTable_stack[-1].addEntry(class_entry)55        self.SymbolTable_stack.append(class_entry.link)56    def END_CLASS(self):57        print 'ending class.'58        self.SymbolTable_stack.pop()59        self.level -= 160    def ENTRY_TYPE(self):61        print "buffering entryType"62        self.SymbolTable_stack.append(self.prevToken_buffer)63        #self.entry_buffer.type = self.prevToken_buffer64    def ENTRY_NAME(self):65        print "buffering entryName"66        self.SymbolTable_stack.append(self.prevToken_buffer)67        #self.entry_buffer.name = self.prevToken_buffer68    def FUNC_ENTRY_TABLE(self):69        print "create_funcEntryAndTable."70        # create a new global/local table entry and link it to the new class table71        #entry = Entry(self.level, self.SymbolTable_stack.pop(), 'function', self.SymbolTable_stack.pop())72        self.entry_buffer.level = self.level73        self.entry_buffer.kind = 'function'74        self.level += 175        self.entry_buffer.link = Symbol_Table(self.level, self.entry_buffer.name)76        # append the new entry to the global table and put the reference to the class table on top the stack77        entry = copy.deepcopy(self.entry_buffer)78        self.SymbolTable_stack[-1].entries.append(entry)79        self.SymbolTable_stack.append(entry.link)80        self.clearEntryBuffer()81    def END_CLASS_FUNC(self):82        print 'ending class function'83        self.SymbolTable_stack.pop()84        self.level -= 185    def ADD_DECL_ARRAY_DIM(self):86        print 'adding an array decl dimension size'87        self.entry_buffer.arraySize.append(int(self.prevToken_buffer))88    def ADD_FUNC_PARAM_ENTRY(self):89        print 'adding a function parameter entry.'90    def CLASS_VAR_ENTRY(self):91        print "creating create_varEntry."92        # create a new class table entry and link it to the new class table93        self.entry_buffer.level = self.level94        self.entry_buffer.kind = 'variable'95        # append the new entry to the global table and put the reference to the class table on top the stack96        entry = copy.deepcopy(self.entry_buffer)97        self.SymbolTable_stack[-1].entries.append(entry)98        self.clearEntryBuffer()99    def PROGRAM_FUNC_ENTRY_TABLE(self):100        print 'adding the program function entry and symbol_table.'101        # create a new global/local table entry and link it to the new class table102        self.entry_buffer.level = self.level103        self.entry_buffer.name = 'program'104        self.entry_buffer.kind = 'function'105        self.level += 1106        self.entry_buffer.link = Symbol_Table(self.level, self.entry_buffer.name)107        # append the new entry to the global table and put the reference to the class table on top the stack108        entry = copy.deepcopy(self.entry_buffer)109        self.SymbolTable_stack[-1].entries.append(entry)110        self.SymbolTable_stack.append(entry.link)...

create_global_table.py

Source:create_global_table.py

...4import boto35import table_info as ti6ddb_eu_1 = boto3.resource('dynamodb', region_name='eu-west-1')7ddb_eu_2 = boto3.resource('dynamodb', region_name='eu-west-2')8def create_global_table(tablename) :9	print("Creating new table called:", tablename)10	for rno in [1,2] :11		region_resource = ddb_eu_1 if rno == 1 else ddb_eu_212		if ti.table_exists_in_region(region_resource, tablename) :13			print("Table already exists in region", rno)14		else :15			print("Creating local table in region", rno)16			t1 = region_resource.create_table(17				TableName = tablename,18				AttributeDefinitions = [19					{ 'AttributeName' : 'kcol', 'AttributeType' : 'S'}20				],21				KeySchema = [22					{ 'AttributeName' : 'kcol', 'KeyType' : 'HASH'}23				],24				StreamSpecification={25					'StreamEnabled' : True,26					'StreamViewType' : 'NEW_AND_OLD_IMAGES'27				},28				BillingMode = 'PAY_PER_REQUEST'		# Default is 'PROVISIIONED, must then specify Read and Write Capacity Units29			)30			print(t1)31			print(t1.attribute_definitions)32			print('Waiting for table to exist...')33			t1.meta.client.get_waiter('table_exists').wait(TableName=tablename)34			print('... finished waiting')35	# Now create global table in the two regions, from the two separate empty tables36	print("Creating global table")37	tgresponse = ddb_eu_2.meta.client.create_global_table(38		GlobalTableName = tablename,39		ReplicationGroup = [ { 'RegionName' : 'eu-west-1' }, { 'RegionName' : 'eu-west-2' } ]40	)41	print("Created global table", tablename)42	print(tgresponse)43if __name__ == "__main__" :44	print(ddb_eu_1)45	print(ddb_eu_2)46	test_session = boto3.session.Session()47	print()48	create_global_table('ddb_global_test_table1')...

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.