PrivacyScanData/target/S3/s3transfer-develop/s3transfer/crt_4.py

def create_s3_crt_client(
    region,
    botocore_credential_provider=None,
    num_threads=None,
    target_throughput=5 * GB / 8,
    part_size=8 * MB,
    use_ssl=True,
    verify=None,
):
    """
    :type region: str
    :param region: The region used for signing

    :type botocore_credential_provider:
        Optional[botocore.credentials.CredentialResolver]
    :param botocore_credential_provider: Provide credentials for CRT
        to sign the request if not set, the request will not be signed

    :type num_threads: Optional[int]
    :param num_threads: Number of worker threads generated. Default
        is the number of processors in the machine.

    :type target_throughput: Optional[int]
    :param target_throughput: Throughput target in Bytes.
        Default is 0.625 GB/s (which translates to 5 Gb/s).

    :type part_size: Optional[int]
    :param part_size: Size, in Bytes, of parts that files will be downloaded
        or uploaded in.

    :type use_ssl: boolean
    :param use_ssl: Whether or not to use SSL.  By default, SSL is used.
        Note that not all services support non-ssl connections.

    :type verify: Optional[boolean/string]
    :param verify: Whether or not to verify SSL certificates.
        By default SSL certificates are verified.  You can provide the
        following values:

        * False - do not validate SSL certificates.  SSL will still be
            used (unless use_ssl is False), but SSL certificates
            will not be verified.
        * path/to/cert/bundle.pem - A filename of the CA cert bundle to
            use. Specify this argument if you want to use a custom CA cert
            bundle instead of the default one on your system.
    """

    event_loop_group = EventLoopGroup(num_threads)
    host_resolver = DefaultHostResolver(event_loop_group)
    bootstrap = ClientBootstrap(event_loop_group, host_resolver)
    provider = None
    tls_connection_options = None

    tls_mode = (
        S3RequestTlsMode.ENABLED if use_ssl else S3RequestTlsMode.DISABLED
    )
    if verify is not None:
        tls_ctx_options = TlsContextOptions()
        if verify:
            tls_ctx_options.override_default_trust_store_from_path(
                ca_filepath=verify
            )
        else:
            tls_ctx_options.verify_peer = False
        client_tls_option = ClientTlsContext(tls_ctx_options)
        tls_connection_options = client_tls_option.new_connection_options()
    if botocore_credential_provider:
        credentails_provider_adapter = CRTCredentialProviderAdapter(
            botocore_credential_provider
        )
        provider = AwsCredentialsProvider.new_delegate(
            credentails_provider_adapter
        )

    target_gbps = target_throughput * 8 / GB
    return S3Client(
        bootstrap=bootstrap,
        region=region,
        credential_provider=provider,
        part_size=part_size,
        tls_mode=tls_mode,
        tls_connection_options=tls_connection_options,
        throughput_target_gbps=target_gbps,
    )