How to use format_timestamp_for_event_source_mapping method in localstack

Best Python code snippet using localstack_python

lambda_api.py

Source:lambda_api.py

...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)247    return {}248@synchronized(lock=EXEC_MUTEX)249def use_docker():250    global DO_USE_DOCKER...

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.