def stream_kwargs_from_args()

in src/aws_encryption_sdk_cli/__init__.py [0:0]

• 19 lines of code
• 11 McCabe index (conditional complexity)

def stream_kwargs_from_args(args, crypto_materials_manager):
    # type: (Namespace, CryptoMaterialsManager) -> STREAM_KWARGS
    """Builds kwargs object for aws_encryption_sdk.stream based on argparse
    arguments and existing CryptoMaterialsManager.

    :param args: Parsed arguments from argparse
    :type args: argparse.Namespace
    :param crypto_materials_manager: Existing CryptoMaterialsManager
    :type crypto_materials_manager: aws_encryption_sdk.materials_manager.base.CryptoMaterialsManager
    :returns: Translated kwargs object for aws_encryption_sdk.stream
    :rtype: dict
    """
    stream_args = {"materials_manager": crypto_materials_manager, "mode": args.action}
    # Look for additional arguments only if encrypting
    if args.action == "encrypt":
        stream_args["encryption_context"] = args.encryption_context
        if args.algorithm is not None:
            stream_args["algorithm"] = getattr(aws_encryption_sdk.Algorithm, args.algorithm)
        if args.frame_length is not None:
            stream_args["frame_length"] = args.frame_length

    if args.commitment_policy is None:
        stream_args["commitment_policy"] = CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
    elif args.commitment_policy == "require-encrypt-require-decrypt":
        stream_args["commitment_policy"] = CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
    elif args.commitment_policy == "require-encrypt-allow-decrypt":
        stream_args["commitment_policy"] = CommitmentPolicy.REQUIRE_ENCRYPT_ALLOW_DECRYPT
    elif args.commitment_policy == "forbid-encrypt-allow-decrypt":
        stream_args["commitment_policy"] = CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT

    if args.max_length is not None:
        stream_args["max_body_length"] = args.max_length
    return stream_args