How to use check_batch_size_range method in localstack

Best Python code snippet using localstack_python

lambda_api.py

Source:lambda_api.py

...161        return details162    if details.qualifier_exists(qualifier):163        return "{}:{}".format(arn, qualifier)164    return arn165def check_batch_size_range(source_arn, batch_size=None):166    source = source_arn.split(":")[2].lower()167    source = "kafka" if "secretsmanager" in source else source168    batch_size_entry = BATCH_SIZE_RANGES.get(source)169    if not batch_size_entry:170        raise ValueError(INVALID_PARAMETER_VALUE_EXCEPTION, "Unsupported event source type")171    batch_size = batch_size or batch_size_entry[0]172    if batch_size > batch_size_entry[1]:173        raise ValueError(174            INVALID_PARAMETER_VALUE_EXCEPTION,175            "BatchSize {} exceeds the max of {}".format(batch_size, batch_size_entry[1]),176        )177    return batch_size178def build_mapping_obj(data) -> Dict:179    mapping = {}180    function_name = data["FunctionName"]181    enabled = data.get("Enabled", True)182    batch_size = data.get("BatchSize")183    mapping["UUID"] = str(uuid.uuid4())184    mapping["FunctionArn"] = func_arn(function_name)185    mapping["LastProcessingResult"] = "OK"186    mapping["StateTransitionReason"] = "User action"187    mapping["LastModified"] = format_timestamp_for_event_source_mapping()188    mapping["State"] = "Enabled" if enabled in [True, None] else "Disabled"189    mapping["ParallelizationFactor"] = data.get("ParallelizationFactor") or 1190    mapping["Topics"] = data.get("Topics") or []191    mapping["MaximumRetryAttempts"] = data.get("MaximumRetryAttempts") or -1192    if "SelfManagedEventSource" in data:193        source_arn = data["SourceAccessConfigurations"][0]["URI"]194        mapping["SelfManagedEventSource"] = data["SelfManagedEventSource"]195        mapping["SourceAccessConfigurations"] = data["SourceAccessConfigurations"]196    else:197        source_arn = data["EventSourceArn"]198        mapping["EventSourceArn"] = source_arn199        mapping["StartingPosition"] = data.get("StartingPosition") or "LATEST"200    batch_size = check_batch_size_range(source_arn, batch_size)201    mapping["BatchSize"] = batch_size202    if data.get("DestinationConfig"):203        mapping["DestinationConfig"] = data.get("DestinationConfig")204    return mapping205def format_timestamp(timestamp=None):206    timestamp = timestamp or datetime.utcnow()207    return isoformat_milliseconds(timestamp) + "+0000"208def format_timestamp_for_event_source_mapping():209    # event source mappings seem to use a different time format (required for Terraform compat.)210    return datetime.utcnow().timestamp()211def add_event_source(data):212    region = LambdaRegion.get()213    mapping = build_mapping_obj(data)214    region.event_source_mappings.append(mapping)215    EventSourceListener.start_listeners(mapping)216    return mapping217def update_event_source(uuid_value, data):218    region = LambdaRegion.get()219    function_name = data.get("FunctionName") or ""220    enabled = data.get("Enabled", True)221    for mapping in region.event_source_mappings:222        if uuid_value == mapping["UUID"]:223            if function_name:224                mapping["FunctionArn"] = func_arn(function_name)225            batch_size = data.get("BatchSize")226            if "SelfManagedEventSource" in mapping:227                batch_size = check_batch_size_range(228                    mapping["SourceAccessConfigurations"][0]["URI"],229                    batch_size or mapping["BatchSize"],230                )231            else:232                batch_size = check_batch_size_range(233                    mapping["EventSourceArn"], batch_size or mapping["BatchSize"]234                )235            mapping["State"] = "Enabled" if enabled in [True, None] else "Disabled"236            mapping["LastModified"] = format_timestamp_for_event_source_mapping()237            mapping["BatchSize"] = batch_size238            if "SourceAccessConfigurations" in (mapping and data):239                mapping["SourceAccessConfigurations"] = data["SourceAccessConfigurations"]240            return mapping241    return {}242def delete_event_source(uuid_value):243    region = LambdaRegion.get()244    for i, m in enumerate(region.event_source_mappings):245        if uuid_value == m["UUID"]:246            return region.event_source_mappings.pop(i)...

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.