def invoke()

in src/rpdk/core/invoke.py [0:0]

• 53 lines of code
• 5 McCabe index (conditional complexity)

def invoke(args):
    _validate_sam_args(args)
    project = Project()
    project.load()

    client = ResourceClient(
        args.function_name,
        args.endpoint,
        args.region,
        project.schema,
        {},
        executable_entrypoint=project.executable_entrypoint,
        docker_image=args.docker_image,
    )

    action = Action[args.action]
    try:
        request = json.load(args.request)
    except ValueError as e:
        raise SysExitRecommendedError(f"Invalid JSON: {e}") from e
    payload = client._make_payload(
        action,
        request["desiredResourceState"],
        request["previousResourceState"],
        request.get("typeConfiguration"),
    )

    current_invocation = 0

    try:
        while _needs_reinvocation(args.max_reinvoke, current_invocation):
            print("=== Handler input ===")

            payload_to_log = {
                "callbackContext": payload["callbackContext"],
                "action": payload["action"],
                "requestData": {
                    "resourceProperties": payload["requestData"]["resourceProperties"],
                    "previousResourceProperties": payload["requestData"][
                        "previousResourceProperties"
                    ],
                },
                "region": payload["region"],
                "awsAccountId": payload["awsAccountId"],
                "bearerToken": payload["bearerToken"],
            }
            print(json.dumps({**payload_to_log}, indent=2))

            response = client._call(payload)
            current_invocation = current_invocation + 1
            print("=== Handler response ===")
            print(json.dumps(response, indent=2))
            status = OperationStatus[response["status"]]

            if status != OperationStatus.IN_PROGRESS:
                break

            sleep(response.get("callbackDelaySeconds", 0))
            payload["callbackContext"] = response.get("callbackContext")
    except KeyboardInterrupt:
        pass