Main

CLI entry point and main execution logic for the Lambda handler.

---

handle

handle(event, context)

Route and execute a Lambda handler function invocation.

This function acts as a router that deserializes the incoming event, loads the appropriate handler, and invokes it with the event payload.

Parameters:

Name	Type	Description	Default
event	LambdaEvent	The Lambda event containing the handler reference and payload. Must be a dictionary with handler and event fields.	required
context	LambdaContext	The AWS Lambda context object providing runtime information.	required

Returns:

Type	Description
Optional[JSON]	The response from the invoked handler, or None if no response.

Raises:

Type	Description
ValueError	If the event is not a dictionary type.

Source code in src/aibs_informatics_aws_lambda/main.py

@logger.inject_lambda_context(log_event=True)
def handle(event: LambdaEvent, context: LambdaContext) -> Optional[JSON]:
    """Route and execute a Lambda handler function invocation.

    This function acts as a router that deserializes the incoming event,
    loads the appropriate handler, and invokes it with the event payload.

    Args:
        event: The Lambda event containing the handler reference and payload.
            Must be a dictionary with handler and event fields.
        context: The AWS Lambda context object providing runtime information.

    Returns:
        The response from the invoked handler, or None if no response.

    Raises:
        ValueError: If the event is not a dictionary type.
    """
    logger.info("Parsing event")
    if not isinstance(event, dict):
        raise ValueError("Unable to parse event - events must be Dict type")
    request = LambdaHandlerRequest.from_dict(event)
    logger.info(f"Successfully deserialized request. Loading handler from '{request.handler}'")
    target_handler = request.handler
    logger.info("Successfully loaded handler. Invoking..")
    response = target_handler(request.event, context)
    logger.info("Invocation is complete. Returning result")
    return response

handle_cli

handle_cli(args=None)

Execute a Lambda handler from the command line interface.

Parses command line arguments to determine the handler to invoke, the event payload, and optionally where to store the response. Supports loading payloads from JSON strings, local files, or S3.

Parameters:

Name	Type	Description	Default
args	Optional[Sequence[str]]	Optional sequence of command line arguments. If None, uses sys.argv.	None

Raises:

Type	Description
ValueError	If handler or payload cannot be resolved from arguments or environment variables.
ValueError	If response location is specified but invalid.

Example

handle_cli([
    '--handler', 'my_module.MyHandler',
    '--payload', '{"key": "value"}',
    '--response-location', '/tmp/response.json'
])

Source code in src/aibs_informatics_aws_lambda/main.py

def handle_cli(args: Optional[Sequence[str]] = None):
    """Execute a Lambda handler from the command line interface.

    Parses command line arguments to determine the handler to invoke,
    the event payload, and optionally where to store the response.
    Supports loading payloads from JSON strings, local files, or S3.

    Args:
        args: Optional sequence of command line arguments. If None,
            uses sys.argv.

    Raises:
        ValueError: If handler or payload cannot be resolved from
            arguments or environment variables.
        ValueError: If response location is specified but invalid.

    Example:
        ```python
        handle_cli([
            '--handler', 'my_module.MyHandler',
            '--payload', '{"key": "value"}',
            '--response-location', '/tmp/response.json'
        ])
        ```
    """
    logger.info("Processing request from CLI")

    parser = argparse.ArgumentParser(description="CLI AWS Lambda Handler")
    parser.add_argument(
        "--handler-qualified-name",
        "--handler-name",
        "--handler",
        dest="handler_qualified_name",
        required=False,
        default=get_env_var(
            AWS_LAMBDA_FUNCTION_HANDLER_KEY, AWS_LAMBDA_FUNCTION_HANDLER_STANDARD_KEY
        ),
        help=(
            f"handler function qualified name. If not provided, will try to load from "
            f"{(AWS_LAMBDA_FUNCTION_HANDLER_KEY, AWS_LAMBDA_FUNCTION_HANDLER_STANDARD_KEY)} "
            "env variables"
        ),
    )
    parser.add_argument(
        "--payload",
        "--event",
        "-e",
        required=False,
        default=get_env_var(AWS_LAMBDA_EVENT_PAYLOAD_KEY),
        help=(
            f"event payload of function. If not provided, will try to load from "
            f"{AWS_LAMBDA_EVENT_PAYLOAD_KEY} env variable"
        ),
    )
    parser.add_argument(
        "--response-location",
        "-o",
        dest="response_location",
        required=False,
        default=get_env_var(AWS_LAMBDA_EVENT_RESPONSE_LOCATION_KEY),
        help=(
            f"optional response location to store response at. can be S3 or local file. If not "
            f"provided, will load from {AWS_LAMBDA_EVENT_RESPONSE_LOCATION_KEY} env variable."
        ),
    )

    logger.info("Parsing arguments")

    parsed_args = parser.parse_args(args=args)

    if parsed_args.handler_qualified_name is None:
        raise ValueError("Handler could not be resolved from argument or environment variable")
    target_handler = deserialize_handler(parsed_args.handler_qualified_name)

    if parsed_args.payload is None:
        raise ValueError("Payload could not be resolved by argument or environment variable")
    if S3URI.is_valid(parsed_args.payload):
        logger.info("Payload is an S3 path. downloading to JSON")
        event = download_to_json_object(S3URI(parsed_args.payload))
    else:
        event = load_json_object(parsed_args.payload)

    logger.info("Successfully loaded handler. Invoking..")
    response = target_handler(event, DefaultLambdaContext())
    logger.info("Invocation complete.")

    response_location = parsed_args.response_location
    if response_location:
        response = response or {}
        logger.info(f"Response location specified: {response_location}")
        if S3URI.is_valid(response_location):
            logger.info("Uploading result to S3")
            upload_json(response, S3URI(response_location))
        elif not os.path.isdir(response_location) and os.access(
            os.path.dirname(response_location), os.W_OK
        ):
            logger.info("Writing result locally")
            with open(response_location, "w") as f:
                json.dump(response, f, sort_keys=True)
        else:
            raise ValueError(
                f"Response location specified {response_location}, but not a valid s3/local path"
            )
    else:
        logger.info("Response location NOT specified. Response not saved.")

    logger.info("Handler execution complete.")