Best Python code snippet using localstack_python
lambda_utils.py
Source:lambda_utils.py
...255 LOG.warn(256 f"Could not convert filter value {numeric_filter[idx + 1]} to a valid number value for filtering"257 )258 return True259def verify_dict_filter(record_value: any, dict_filter: Dict[str, any]) -> bool:260 # https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html#filtering-syntax261 fits_filter = False262 for key, filter_value in dict_filter.items():263 if key.lower() == "anything-but":264 fits_filter = record_value not in filter_value265 elif key.lower() == "numeric":266 fits_filter = parse_and_apply_numeric_filter(record_value, filter_value)267 elif key.lower() == "exists":268 fits_filter = bool(filter_value) # exists means that the key exists in the event record269 elif key.lower() == "prefix":270 if not isinstance(record_value, str):271 LOG.warn(f"Record Value {record_value} does not seem to be a valid string.")272 fits_filter = isinstance(record_value, str) and record_value.startswith(273 str(filter_value)274 )275 if fits_filter:276 return True277 return fits_filter278def filter_stream_record(filter_rule: Dict[str, any], record: Dict[str, any]) -> bool:279 if not filter_rule:280 return True281 # https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html#filtering-syntax282 filter_results = []283 for key, value in filter_rule.items():284 # check if rule exists in event285 record_value = (286 record.get(key.lower(), record.get(key)) if isinstance(record, Dict) else None287 )288 append_record = False289 if record_value is not None:290 # check if filter rule value is a list (leaf of rule tree) or a dict (rescursively call function)291 if isinstance(value, list):292 if len(value) > 0:293 if isinstance(value[0], (str, int)):294 append_record = record_value in value295 if isinstance(value[0], dict):296 append_record = verify_dict_filter(record_value, value[0])297 else:298 LOG.warn(f"Empty lambda filter: {key}")299 elif isinstance(value, dict):300 append_record = filter_stream_record(value, record_value)301 else:302 # special case 'exists'303 if isinstance(value, list) and len(value) > 0:304 append_record = not value[0].get("exists", True)305 filter_results.append(append_record)306 return all(filter_results)307def filter_stream_records(records, filters: List[FilterCriteria]):308 filtered_records = []309 for record in records:310 for filter in filters:...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!