AutoAndroidController/py/venv/Lib/site-packages/baidubce/services/vpc/vpc_client.py

# Copyright (c) 2014 Baidu.com, Inc. All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
# except in compliance with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the
# License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
# either express or implied. See the License for the specific language governing permissions
# and limitations under the License.

"""
This module provides a client class for VPC.
"""

import copy
import json
import logging
import uuid

from baidubce import bce_base_client
from baidubce.auth import bce_v1_signer
from baidubce.http import bce_http_client
from baidubce.http import handler
from baidubce.http import http_methods

from baidubce.utils import required
from baidubce import compat

_logger = logging.getLogger(__name__)


class VpcClient(bce_base_client.BceBaseClient):
    """
    VPC base sdk client
    """

    prefix = b'/v1'

    def __init__(self, config=None):
        bce_base_client.BceBaseClient.__init__(self, config)

    def _merge_config(self, config=None):
        """
        :param config:
        :type config: baidubce.BceClientConfiguration
        :return:
        """
        if config is None:
            return self.config
        else:
            new_config = copy.copy(self.config)
            new_config.merge_non_none_values(config)
            return new_config

    def _send_request(self, http_method, path,
                      body=None, headers=None, params=None,
                      config=None, body_parser=None):
        config = self._merge_config(config)
        if body_parser is None:
            body_parser = handler.parse_json
        if headers is None:
            headers = {b'Accept': b'*/*', b'Content-Type': b'application/json;charset=utf-8'}
        return bce_http_client.send_request(
            config, bce_v1_signer.sign, [handler.parse_error, body_parser],
            http_method, VpcClient.prefix + path, body, headers, params)

    @required(name=(bytes, str), cidr=(bytes, str))
    def create_vpc(self, name, cidr, description=None, client_token=None, enable_ipv6=None, tags=None, config=None):
        """
        The name of vpc to be created.

        :param name:
            The name of vpc to be created.
        :type name: string

        :param cidr:
            The CIDR of the vpc.
        :type cidr: string

        :param description:
            The description of the vpc.
        :type description: string

        :param enable_ipv6:
            The option param to enable or disable ipv6 for the subnet
        :type enable_ipv6: boolean

        :param tags:
            List of tags to be bind
        :type tags: list

        :param client_token:
            An ASCII string whose length is less than 64.
            The request will be idempotent if clientToken is provided.
            If the clientToken is not specified by the user, a random String generated by default algorithm will be used.
        :type client_token: string

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc'
        params = {}
        if client_token is None:
            params[b'clientToken'] = generate_client_token()
        else:
            params[b'clientToken'] = client_token

        body = {
            'name': compat.convert_to_string(name),
            'cidr': compat.convert_to_string(cidr)
        }
        if description is not None:
            body['description'] = compat.convert_to_string(description)

        if tags is not None:
            body['tags'] = tags

        if enable_ipv6 is not None:
            body['enableIpv6'] = enable_ipv6

        return self._send_request(http_methods.POST, path, body=json.dumps(body), params=params,
                                  config=config)

    @required(marker=(bytes, str), max_Keys=int, is_Default=bool)
    def list_vpcs(self, marker=None, max_Keys=None, vpc_ids=None, isDefault=None, config=None):
        """
        Return a list of vpcs owned by the authenticated user.

        :param marker:
            The optional parameter marker specified in the original request to specify
            where in the results to begin listing.
            Together with the marker, specifies the list result which listing should begin.
            If the marker is not specified, the list result will listing from the first one.
        :type marker: string

        :param max_keys:
            The optional parameter to specifies the max number of list result to return.
            The default value is 1000.
        :type max_keys: int

        :param vpc_ids:
            filter by vpc_ids, optional parameter
        :type vpc_ids: string

        :param isDefault:
            The option param demotes whether the vpc is default vpc.
        :type isDefault: boolean

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc'
        params = {}

        if marker is not None:
            params[b'marker'] = marker
        if max_Keys is not None:
            params[b'maxKeys'] = max_Keys
        if vpc_ids is not None:
            params[b'vpcIds'] = vpc_ids
        if isDefault is not None:
            params[b'isDefault'] = isDefault

        return self._send_request(http_methods.GET, path, params=params, config=config)

    @required(vpc_id=(bytes, str))
    def get_vpc(self, vpc_id, config=None):
        """
        Get the detail information of specified vpc.

        :param vpc_id:
            The id of vpc.
        :type vpc_id: string

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc/%s' % compat.convert_to_bytes(vpc_id)

        return self._send_request(http_methods.GET, path, config=config)

    @required(vpc_id=(bytes, str))
    def delete_vpc(self, vpc_id, client_token=None, config=None):
        """
        Delete the specified vpc owned by the user.All resource in the vpc must be deleted before the vpc itself
        can be deleted.

        :param vpc_id:
            The id of instance.
        :type vpc_id: string

        :param client_token:
            An ASCII string whose length is less than 64.
            The request will be idempotent if clientToken is provided.
            If the clientToken is not specified by the user, a random String generated by default algorithm will
            be used.
        :type client_token: string

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc/%s' % compat.convert_to_bytes(vpc_id)
        params = {}

        if client_token is None:
            params[b'clientToken'] = generate_client_token()
        else:
            params[b'clientToken'] = client_token

        return self._send_request(http_methods.DELETE, path, params=params, config=config)

    @required(vpc_id=(bytes, str), name=(bytes, str))
    def update_vpc(self, vpc_id, name, description=None, enable_ipv6=None,
                          secondaryCidr=None, client_token=None, config=None):
        """
        Modify the special attribute to new value of the vpc owned by the user.

        :param vpc_id:
            The id of the specified vpc
        :type vpc_id: string

        :param name:
            The name of the specified vpc
        :type name: string

        :param description:
           The description of the vpc.
        :type description: string

        :param enable_ipv6:
            The option param to enable or disable ipv6 for the subnet
        :type enable_ipv6: boolean

        :param secondaryCidr:
            This parameter specifies the secondary CIDR ranges associated with the VPC.
        :type secondaryCidr: list

        :param client_token:
            An ASCII string whose length is less than 64.
            The request will be idempotent if clientToken is provided.
            If the clientToken is not specified by the user, a random String generated by default algorithm will be used.
        :type client_token: string

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """

        path = b'/vpc/%s' % compat.convert_to_bytes(vpc_id)
        params = {
            b'modifyAttribute': None
        }
        body = {
            'name': compat.convert_to_string(name)
        }

        if client_token is None:
            params[b'clientToken'] = generate_client_token()
        else:
            params[b'clientToken'] = client_token

        if description is not None:
            body['description'] = compat.convert_to_string(description)

        if enable_ipv6 is not None:
            body['enableIpv6'] = enable_ipv6

        if secondaryCidr is not None:
            body['secondaryCidr'] = secondaryCidr

        return self._send_request(http_methods.PUT, path, json.dumps(body),
                                  params=params, config=config)

    @required(vpc_id=(bytes, str))
    def get_private_ip_address_info(self, vpc_id, private_ip_range=None, private_ip_addresses=None, config=None):
        """
        Get the privateIpAddressesInfo from vpc.

        :param vpc_id:
            The id of vpc.
        :type vpc_id: string

        :param private_ip_range:
            The optional parameter specifying the private IP range.
        :type private_ip_range: string

        :param private_ip_addresses:
            The optional list parameter specifying the private IP addresses.
        :type private_ip_addresses: list

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc/%s/privateIpAddressInfo' % compat.convert_to_bytes(vpc_id)
        params = {}

        if private_ip_range is not None:
            params[b'privateIpRange'] = private_ip_range

        if private_ip_addresses is not None:
            params[b'privateIpAddresses'] = (",").join(private_ip_addresses)

        return self._send_request(http_methods.GET, path, params=params, config=config)

    @required(vpc_id=(bytes, str))
    def open_relay(self, vpc_id, client_token=None, config=None):
        """
        Open relay for the specified vpc.

        :param vpc_id:
            The id of vpc.
        :type vpc_id: string

        :param client_token:
            An ASCII string whose length is less than 64.
            The request will be idempotent if clientToken is provided.
            If the clientToken is not specified by the user, a random String generated by default algorithm will be used.
        :type client_token: string

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc/openRelay/%s' % compat.convert_to_bytes(vpc_id)
        params = {}

        if client_token is None:
            params[b'clientToken'] = generate_client_token()
        else:
            params[b'clientToken'] = client_token

        return self._send_request(http_methods.PUT, path, config=config)

    @required(vpc_id=(bytes, str))
    def shutdown_relay(self, vpc_id, client_token=None, config=None):
        """
        Shutdown relay for the specified vpc.

        :param vpc_id:
            The id of vpc.
        :type vpc_id: string

        :param client_token:
            An ASCII string whose length is less than 64.
            The request will be idempotent if clientToken is provided.
            If the clientToken is not specified by the user, a random String generated by default algorithm will be used.
        :type client_token: string

        :param config:
        :type config: baidubce.BceClientConfiguration

        :return:
        :rtype baidubce.bce_response.BceResponse
        """
        path = b'/vpc/shutdownRelay/%s' % compat.convert_to_bytes(vpc_id)
        params = {}
        
        if client_token is None:
            params[b'clientToken'] = generate_client_token()
        else:
            params[b'clientToken'] = client_token

        return self._send_request(http_methods.PUT, path, config=config)

def generate_client_token_by_uuid():
    """
    The default method to generate the random string for client_token
    if the optional parameter client_token is not specified by the user.

    :return:
    :rtype string
    """
    return str(uuid.uuid4())


generate_client_token = generate_client_token_by_uuid