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

How to use get_key_schema method in localstack

Best Python code snippet using localstack_python

message_schemas.py

Source:message_schemas.py

...3from . import config4# The SQL query that pulls the query stats also includes the plan_handle as part of its grouping key. However, for the5# Kafka messages we want to ensure that all plans for the same sql_handle are in the same Kafka partition so that they6# are handled by the same detect-step consumer, so the plan_handle is not included here:7def get_key_schema(record_name: str) -> str:8    return json.dumps({9        "name": record_name,10        "namespace": config.AVRO_SCHEMA_NAMESPACE,11        "type": "record",12        "fields": [13            {14                "name": "db_identifier",15                "type": "string"16            },17            {18                "name": "set_options",19                "type": "int"20            },21            {22                "name": "sql_handle",23                "type": "string"24            }25        ]26    })27def key_from_value(message_value: Dict[str, Any]) -> Dict[str, Any]:28    return {"db_identifier": message_value["db_identifier"],29            "set_options": message_value["set_options"],30            "sql_handle": message_value["sql_handle"]}31QUERY_STATS_MESSAGE_KEY_AVRO_SCHEMA = get_key_schema('query_stats_key')32BAD_PLANS_MESSAGE_KEY_AVRO_SCHEMA = get_key_schema('bad_plans_key')33EVICTED_PLANS_MESSAGE_KEY_AVRO_SCHEMA = get_key_schema('evicted_plans_key')34SINGLE_PLAN_STATS_FIELDS = [35    {36        "name": "db_identifier",37        "type": "string"38    },39    {40        "name": "plan_handle",41        "type": "string"42    },43    {44        "name": "sql_handle",45        "type": "string"46    },47    {...

CreateTableOpreationInDynamodbCF.py

Source:CreateTableOpreationInDynamodbCF.py

1import copy2from enum import Enum3from switchlang import switch4from AWSModules.DynamoDbOpreationalModule.DynamoDbOpreationHelper import DynamoDbOpreationHelper5from AWSModules.DynanoDbDatabaseModule.AttributeDefinationsMapper import AttributeDefinationsMapper6from AWSModules.DynanoDbDatabaseModule.CreateTableInDynamodb import CreateTableInDynamoDb7from AWSModules.DynanoDbDatabaseModule.KeySchemaMapper import KeySchemaMapper8from CommonCode.List.List import List9from CommonCode.strings import Strings10from Environment.EnvironmentTypeEnum import EnvironmentTypeEnum11class State(Enum):12    CHECK_TABLE_NAME = 0;13    GET_KEY_SCHEMA = 114    GET_ATTRIBUTE_TYPE = 215    CREATE_TABLE = 316    DONE = 4;17class CreateTableOpreationInDynamodbCF:18    m_TableName = None;19    m_keyType = None20    m_keySchema = None21    m_attributeType = None22    m_createTable = CreateTableInDynamoDb()23    m_dynamoDbOpreationHeper = DynamoDbOpreationHelper()24    m_keySchemaMapper = KeySchemaMapper()25    m_attributeMapper = None26    def __init__(self, keyType, config):27        self.m_keyType = keyType28        self.m_attributeMapper = AttributeDefinationsMapper(config)29    def start(self, tableName):30        self.m_TableName = tableName31        self.controlFlow(currentState=State.CHECK_TABLE_NAME)32    def done(self):33        return34    def checkTableNameIsEmpty(self):35        if (Strings.isEmpty(self.m_TableName)):36            raise Exception("table name is Empty")37        else:38            self.controlFlow(currentState=State.GET_KEY_SCHEMA)39    def keySchema(self):40        self.m_keySchema = copy.copy(self.m_keySchemaMapper.keyMapper(keyType=self.m_keyType))41        if (self.m_keySchema == None):42            raise Exception("Error while Building key Schema")43        else:44            self.controlFlow(currentState=State.GET_ATTRIBUTE_TYPE)45    def attributeType(self):46        self.m_attributeType = self.m_dynamoDbOpreationHeper.getAttributeList(self.m_keySchema,self.m_attributeMapper.attributeMapper(),self.m_keyType)47        if (self.m_attributeType == None):48            raise Exception("Error while Building Attribute")49        else:50            self.controlFlow(currentState=State.CREATE_TABLE)51    def createTable(self):52        for env in EnvironmentTypeEnum:53            self.m_createTable.createTable(54                self.m_dynamoDbOpreationHeper.getTableName(tableName=self.m_TableName, eviornment=env),55                keySchemaList=self.m_keySchema, AttributeDefinitionList=self.m_attributeType)56    def controlFlow(self, currentState):57        with switch(currentState) as s:58            s.case(State.CHECK_TABLE_NAME, self.checkTableNameIsEmpty)59            s.case(State.GET_KEY_SCHEMA, self.keySchema)60            s.case(State.GET_ATTRIBUTE_TYPE, self.attributeType)61            s.case(State.CREATE_TABLE, self.createTable)62            s.case(State.DONE, self.done)...

create_new_table.py

Source:create_new_table.py

...7HASH_COLUMN_NAME_TYPE = 'S'8RANGE_COLUMN_NAME_TYPE = 'S'9READ_CAPACITY_UNITS = 1010WRITE_CAPACITY_UNITS = 1011def get_key_schema(12        hash_column_name: str,13        range_column_name: str) -> List[dict]:14    return [15        {16            'AttributeName': hash_column_name,17            'KeyType': 'HASH'18        },19        {20            'AttributeName': range_column_name,21            'KeyType': 'RANGE'22        },23    ]24def get_attribute_definitions(25        hash_column_name: str,26        hash_column_name_type: str,27        range_column_name: str,28        range_column_name_type: str29) -> List[dict]:30    return [31        {32            'AttributeName': hash_column_name,33            'AttributeType': hash_column_name_type34        },35        {36            'AttributeName': range_column_name,37            'AttributeType': range_column_name_type38        }39    ]40def get_provisioned_throughput(41        read_capacity_units: int,42        write_capacity_units: int) -> dict:43    return {44        'ReadCapacityUnits': read_capacity_units,45        'WriteCapacityUnits': write_capacity_units46    }47def lambda_handler(event, context):48    """49    Creates a new DynamoDB table to50    storage data about cereals.51    """52    dynamo_db_client = boto3.client('dynamodb')53    key_schema = get_key_schema(54        hash_column_name=HASH_COLUMN_NAME,55        range_column_name=RANGE_COLUMN_NAME56        )57    attribute_definitions = get_attribute_definitions(58        hash_column_name=HASH_COLUMN_NAME,59        hash_column_name_type=HASH_COLUMN_NAME_TYPE,60        range_column_name=RANGE_COLUMN_NAME,61        range_column_name_type=RANGE_COLUMN_NAME_TYPE62        )63    provisioned_throughput = get_provisioned_throughput(64        read_capacity_units=READ_CAPACITY_UNITS,65        write_capacity_units=WRITE_CAPACITY_UNITS66        )67    dynamo_db_client.create_table(...

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.