def main()

in aws_lambda_builders/__main__.py [0:0]

• 56 lines of code
• 6 McCabe index (conditional complexity)

def main():  # pylint: disable=too-many-statements
    """
    Implementation of CLI Interface. Handles only one JSON-RPC method at a time and responds with data

    Input is passed as JSON string either through stdin or as the first argument to the command. Output is always
    printed to stdout.
    """

    # For now the request is not validated
    if len(sys.argv) > 1:
        request_str = sys.argv[1]
        LOG.debug("Using the request object from command line argument")
    else:
        LOG.debug("Reading the request object from stdin")
        request_str = sys.stdin.read()

    request = json.loads(request_str)
    request_id = request["id"]
    params = request["params"]

    # Currently, this is the only supported method
    if request["method"] != "LambdaBuilder.build":
        response = _error_response(request_id, -32601, "Method unavailable")
        return _write_response(response, 1)

    try:
        protocol_version = _parse_version(params.get("__protocol_version"))
        version_compatibility_check(protocol_version)

    except ValueError:
        response = _error_response(request_id, 505, "Unsupported Protocol Version")
        return _write_response(response, 1)

    capabilities = params["capability"]
    supported_workflows = params.get("supported_workflows")

    exit_code = 0
    response = None

    try:
        builder = LambdaBuilder(
            language=capabilities["language"],
            dependency_manager=capabilities["dependency_manager"],
            application_framework=capabilities["application_framework"],
            supported_workflows=supported_workflows,
        )

        artifacts_dir = params["artifacts_dir"]
        builder.build(
            params["source_dir"],
            params["artifacts_dir"],
            params["scratch_dir"],
            params["manifest_path"],
            executable_search_paths=params.get("executable_search_paths", None),
            runtime=params["runtime"],
            optimizations=params["optimizations"],
            options=params["options"],
            mode=params.get("mode", None),
            download_dependencies=params.get("download_dependencies", True),
            dependencies_dir=params.get("dependencies_dir", None),
            combine_dependencies=params.get("combine_dependencies", True),
            architecture=params.get("architecture", X86_64),
        )

        # Return a success response
        response = _success_response(request_id, artifacts_dir)

    except (WorkflowNotFoundError, WorkflowUnknownError, WorkflowFailedError) as ex:
        LOG.debug("Builder workflow failed", exc_info=ex)
        exit_code = 1
        response = _error_response(request_id, 400, str(ex))

    except Exception as ex:
        LOG.debug("Builder crashed", exc_info=ex)
        exit_code = 1
        response = _error_response(request_id, 500, str(ex))

    _write_response(response, exit_code)