• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python client.HTTPClient类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中quantumclient.client.HTTPClient的典型用法代码示例。如果您正苦于以下问题:Python HTTPClient类的具体用法?Python HTTPClient怎么用?Python HTTPClient使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了HTTPClient类的18个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: __init__

    def __init__(self, args_str=None):
        self._args = None
        if not args_str:
            args_str = ' '.join(sys.argv[1:])
        self._parse_args(args_str)

        httpclient = HTTPClient(
            username='admin', tenant_name='demo', password='contrail123',
            # region_name=self._region_name,
            auth_url='http://%s:5000/v2.0' % (self._args.api_server_ip))
        httpclient.authenticate()

        #OS_URL = httpclient.endpoint_url
        OS_URL = 'http://%s:9696/' % (self._args.api_server_ip)
        OS_TOKEN = httpclient.auth_token
        self._quantum = client.Client(
            '2.0', endpoint_url=OS_URL, token=OS_TOKEN)

        self._vnc_lib = VncApi(self._args.admin_user,
                               self._args.admin_password,
                               self._args.admin_tenant_name,
                               self._args.api_server_ip,
                               self._args.api_server_port, '/')

        self._create_vn('public', self._args.public_subnet)
        self._policy_link_vns()
开发者ID:Juniper,项目名称:contrail-controller,代码行数:26,代码来源:demo_cfg.py


示例2: __init__

    def __init__(self, project, user, passwd, api_server_ip):
        AUTH_URL = 'http://%s:5000/v2.0' % (api_server_ip)
        httpclient = HTTPClient(username=user, tenant_name=project,
                                password=passwd, auth_url=AUTH_URL)
        httpclient.authenticate()

        OS_URL = 'http://%s:9696/' % (api_server_ip)
        OS_TOKEN = httpclient.auth_token
        self._quantum = client.Client('2.0', endpoint_url=OS_URL,
                                      token=OS_TOKEN)
开发者ID:Doude,项目名称:contrail-controller,代码行数:10,代码来源:network.py


示例3: _do_quantum_authentication

 def _do_quantum_authentication(self):
     try:
         httpclient = HTTPClient(username=self.username,
                                 tenant_id= self.project_id,
                                 password=self.password,
                                 auth_url=self.auth_url)
         httpclient.authenticate()
     except CommonNetworkClientException, e:
         self.logger.exception('Exception while connection to Quantum')
         raise e
开发者ID:ruchi123,项目名称:contrail-test,代码行数:10,代码来源:quantum_test.py


示例4: initialize

 def initialize(self):
     if not self._url:
         httpclient = HTTPClient(username=self._username,
                                 tenant_name=self._tenant_name,
                                 password=self._password,
                                 region_name=self._region_name,
                                 auth_url=self._auth_url)
         httpclient.authenticate()
         # Populate other password flow attributes
         self._token = httpclient.auth_token
         self._url = httpclient.endpoint_url
开发者ID:akushwah,项目名称:python-quantumclient-n1kv,代码行数:11,代码来源:clientmanager.py


示例5: setUp

 def setUp(self):
     super(QuantumFixture, self).setUp()
     project_id = get_plain_uuid(self.project_id)
     try:
         httpclient = HTTPClient(username=self.username,
                                 tenant_id= project_id,
                                 password=self.password,
                                 auth_url=self.auth_url)
         httpclient.authenticate()
     except CommonNetworkClientException, e:
         self.logger.exception('Exception while connection to Quantum')
         raise e
开发者ID:rombie,项目名称:contrail-test,代码行数:12,代码来源:quantum_test.py


示例6: test_endpoint_type

    def test_endpoint_type(self):
        resources = copy.deepcopy(KS_TOKEN_RESULT)
        endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
        endpoints['internalURL'] = 'internal'
        endpoints['adminURL'] = 'admin'
        endpoints['publicURL'] = 'public'

        # Test default behavior is to choose public.
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION)

        self.client._extract_service_catalog(resources)
        self.assertEqual(self.client.endpoint_url, 'public')

        # Test admin url
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION, endpoint_type='adminURL')

        self.client._extract_service_catalog(resources)
        self.assertEqual(self.client.endpoint_url, 'admin')

        # Test public url
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION, endpoint_type='publicURL')

        self.client._extract_service_catalog(resources)
        self.assertEqual(self.client.endpoint_url, 'public')

        # Test internal url
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION,
                                 endpoint_type='internalURL')

        self.client._extract_service_catalog(resources)
        self.assertEqual(self.client.endpoint_url, 'internal')

        # Test url that isn't found in the service catalog
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION,
                                 endpoint_type='privateURL')

        self.assertRaises(exceptions.EndpointTypeNotFound,
                          self.client._extract_service_catalog,
                          resources)
开发者ID:,项目名称:,代码行数:49,代码来源:


示例7: setUp

 def setUp(self):
     """Prepare the test environment"""
     super(CLITestAuthKeystone, self).setUp()
     self.mox = mox.Mox()
     self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                              password=PASSWORD, auth_url=AUTH_URL,
                              region_name=REGION)
     self.addCleanup(self.mox.VerifyAll)
     self.addCleanup(self.mox.UnsetStubs)
开发者ID:StackOps,项目名称:python-quantumclient,代码行数:9,代码来源:test_auth.py


示例8: __init__

 def __init__(self, **kwargs):
     """ Initialize a new client for the Quantum v2.0 API. """
     super(Client, self).__init__()
     self.httpclient = HTTPClient(**kwargs)
     self.version = '2.0'
     self.format = 'json'
     self.action_prefix = "/v%s" % (self.version)
     self.retries = 0
     self.retry_interval = 1
开发者ID:dani4571,项目名称:python-quantumclient,代码行数:9,代码来源:client.py


示例9: setUp

 def setUp(self):
     super(QuantumFixture, self).setUp()
     project_id = get_plain_uuid(self.project_id)
     insecure = bool(os.getenv('OS_INSECURE', True))
     try:
         httpclient = HTTPClient(username=self.username,
                                 tenant_id=project_id,
                                 password=self.password,
                                 auth_url=self.auth_url,
                                 insecure=insecure)
         httpclient.authenticate()
     except CommonNetworkClientException as e:
         self.logger.exception('Exception while connection to Quantum')
         raise e
     OS_URL = httpclient.endpoint_url
     OS_TOKEN = httpclient.auth_token
     self.obj = client.Client(
         '2.0',
         endpoint_url=OS_URL,
         token=OS_TOKEN,
         insecure=insecure)
     self.project_id = httpclient.auth_tenant_id
开发者ID:smusuvat,项目名称:contrail-test,代码行数:22,代码来源:quantum_test.py


示例10: _run

    def _run(self, args_str=None, oper=''):
        self._args = None
        if not args_str:
            args_str = ' '.join(sys.argv[1:])
        self._parse_args(args_str)

        if self._quantum == None:
            httpclient = HTTPClient(username='admin',
                                    tenant_name='demo',
                                    password='contrail123',
                                    # region_name=self._region_name,
                                    auth_url='http://%s:5000/v2.0' % (self._args.api_server_ip))
            httpclient.authenticate()

            #OS_URL = httpclient.endpoint_url
            OS_URL = 'http://%s:9696/' % (self._args.api_server_ip)
            OS_TOKEN = httpclient.auth_token
            self._quantum = client.Client(
                '2.0', endpoint_url=OS_URL, token=OS_TOKEN)

            self._vnc_lib = VncApi(self._args.admin_user,
                                   self._args.admin_password,
                                   self._args.admin_tenant_name,
                                   # self._args.vn_name,
                                   self._args.api_server_ip,
                                   self._args.api_server_port, '/')

            self._proj_obj = self._vnc_lib.project_read(
                fq_name=['default-domain', 'demo'])
            self._ipam_obj = self._vnc_lib.network_ipam_read(
                fq_name=['default-domain', 'default-project', 'default-network-ipam'])

        if self._args.oper == 'add':
            self._create_vn(self._args.vn_name, self._args.public_subnet)
        elif self._args.oper == 'del':
            self._delete_vn(self._args.vn_name)
        elif self._args.oper == 'add-list':
            self._create_vn_list(self._args.vn_list)
开发者ID:alokkumar223,项目名称:contrail-test,代码行数:38,代码来源:vn_oper.py


示例11: test_get_endpoint_url_other

    def test_get_endpoint_url_other(self):
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION, endpoint_type='otherURL')
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200

        self.client.request(StrContains(AUTH_URL +
                                        '/tokens/%s/endpoints' % TOKEN), 'GET',
                            headers=IsA(dict)). \
            AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
        self.mox.ReplayAll()
        self.assertRaises(exceptions.EndpointTypeNotFound,
                          self.client.do_request,
                          '/resource',
                          'GET')
开发者ID:,项目名称:,代码行数:20,代码来源:


示例12: Client


#.........这里部分代码省略.........
    def update_health_monitor(self, health_monitor, body=None):
        """
        Updates a load balancer health monitor
        """
        return self.put(self.health_monitor_path % (health_monitor), body=body)

    @APIParamsCall
    def delete_health_monitor(self, health_monitor):
        """
        Deletes the specified load balancer health monitor
        """
        return self.delete(self.health_monitor_path % (health_monitor))

    @APIParamsCall
    def associate_health_monitor(self, pool, body):
        """
        Associate  specified load balancer health monitor and pool
        """
        return self.post(self.associate_pool_health_monitors_path % (pool),
                         body=body)

    @APIParamsCall
    def disassociate_health_monitor(self, pool, health_monitor):
        """
        Disassociate specified load balancer health monitor and pool
        """
        path = (self.disassociate_pool_health_monitors_path %
                {'pool': pool, 'health_monitor': health_monitor})
        return self.delete(path)

    def __init__(self, **kwargs):
        """ Initialize a new client for the Quantum v2.0 API. """
        super(Client, self).__init__()
        self.httpclient = HTTPClient(**kwargs)
        self.version = '2.0'
        self.format = 'json'
        self.action_prefix = "/v%s" % (self.version)
        self.retries = 0
        self.retry_interval = 1

    def _handle_fault_response(self, status_code, response_body):
        # Create exception with HTTP status code and message
        _logger.debug("Error message: %s", response_body)
        # Add deserialized error message to exception arguments
        try:
            des_error_body = self.deserialize(response_body, status_code)
        except:
            # If unable to deserialized body it is probably not a
            # Quantum error
            des_error_body = {'message': response_body}
        # Raise the appropriate exception
        exception_handler_v20(status_code, des_error_body)

    def do_request(self, method, action, body=None, headers=None, params=None):
        # Add format and tenant_id
        action += ".%s" % self.format
        action = self.action_prefix + action
        if type(params) is dict and params:
            action += '?' + urllib.urlencode(params, doseq=1)
        if body:
            body = self.serialize(body)
        self.httpclient.content_type = self.content_type()
        resp, replybody = self.httpclient.do_request(action, method, body=body)
        status_code = self.get_status_code(resp)
        if status_code in (httplib.OK,
                           httplib.CREATED,
开发者ID:dani4571,项目名称:python-quantumclient,代码行数:67,代码来源:client.py


示例13: Client


#.........这里部分代码省略.........
        return self.get(self.subnets_path, params=_params)

    @APIParamsCall
    def show_subnet(self, subnet, **_params):
        """
        Fetches information of a certain subnet
        """
        return self.get(self.subnet_path % (subnet), params=_params)

    @APIParamsCall
    def create_subnet(self, body=None):
        """
        Creates a new subnet
        """
        return self.post(self.subnets_path, body=body)

    @APIParamsCall
    def update_subnet(self, subnet, body=None):
        """
        Updates a subnet
        """
        return self.put(self.subnet_path % (subnet), body=body)

    @APIParamsCall
    def delete_subnet(self, subnet):
        """
        Deletes the specified subnet
        """
        return self.delete(self.subnet_path % (subnet))

    def __init__(self, **kwargs):
        """ Initialize a new client for the Quantum v2.0 API. """
        super(Client, self).__init__()
        self.httpclient = HTTPClient(**kwargs)
        self.version = '2.0'
        self.format = 'json'
        self.action_prefix = "/v%s" % (self.version)
        self.retries = 0
        self.retry_interval = 1

    def _handle_fault_response(self, status_code, response_body):
        # Create exception with HTTP status code and message
        error_message = response_body
        _logger.debug("Error message: %s", error_message)
        # Add deserialized error message to exception arguments
        try:
            des_error_body = Serializer().deserialize(error_message,
                                                      self.content_type())
        except:
            # If unable to deserialized body it is probably not a
            # Quantum error
            des_error_body = {'message': error_message}
        # Raise the appropriate exception
        exception_handler_v20(status_code, des_error_body)

    def do_request(self, method, action, body=None, headers=None, params=None):
        # Add format and tenant_id
        action += ".%s" % self.format
        action = self.action_prefix + action
        if type(params) is dict:
            action += '?' + urllib.urlencode(params, doseq=1)
        if body:
            body = self.serialize(body)
        self.httpclient.content_type = self.content_type()
        resp, replybody = self.httpclient.do_request(action, method, body=body)
        status_code = self.get_status_code(resp)
开发者ID:codeoedoc,项目名称:python-quantumclient,代码行数:67,代码来源:client.py


示例14: setUp

 def setUp(self):
     """Prepare the test environment"""
     self.mox = mox.Mox()
     self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                              password=PASSWORD, auth_url=AUTH_URL,
                              region_name=REGION)
开发者ID:akushwah,项目名称:python-quantumclient-n1kv,代码行数:6,代码来源:test_auth.py


示例15: CLITestAuthKeystone

class CLITestAuthKeystone(unittest.TestCase):

    def setUp(self):
        """Prepare the test environment"""
        self.mox = mox.Mox()
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION)

    def tearDown(self):
        """Clear the test environment"""
        self.mox.VerifyAll()
        self.mox.UnsetStubs()

    def test_get_token(self):
        self.mox.StubOutWithMock(self.client, "request")

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200

        self.client.request(AUTH_URL + '/tokens', 'POST',
                            body=IsA(str), headers=IsA(dict)).\
            AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res200, ''))
        self.mox.ReplayAll()

        self.client.do_request('/resource', 'GET')
        self.assertEqual(self.client.endpoint_url, ENDPOINT_URL)
        self.assertEqual(self.client.auth_token, TOKEN)
        self.assertEqual(self.client.token_retrieved, True)

    def test_already_token_retrieved(self):
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN
        self.client.endpoint_url = ENDPOINT_URL
        self.client.token_retrieved = True

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200

        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res200, ''))
        self.mox.ReplayAll()

        self.client.do_request('/resource', 'GET')

    def test_refresh_token(self):
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN
        self.client.endpoint_url = ENDPOINT_URL
        self.client.token_retrieved = True

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200
        res401 = self.mox.CreateMock(httplib2.Response)
        res401.status = 401

        # If a token is expired, quantum server retruns 401
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res401, ''))
        self.client.request(AUTH_URL + '/tokens', 'POST',
                            body=IsA(str), headers=IsA(dict)).\
            AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res200, ''))
        self.mox.ReplayAll()
        self.client.do_request('/resource', 'GET')
开发者ID:akushwah,项目名称:python-quantumclient-n1kv,代码行数:74,代码来源:test_auth.py


示例16: CLITestAuthKeystone

class CLITestAuthKeystone(testtools.TestCase):

    def setUp(self):
        """Prepare the test environment."""
        super(CLITestAuthKeystone, self).setUp()
        self.mox = mox.Mox()
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION)
        self.addCleanup(self.mox.VerifyAll)
        self.addCleanup(self.mox.UnsetStubs)

    def test_get_token(self):
        self.mox.StubOutWithMock(self.client, "request")

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200

        self.client.request(AUTH_URL + '/tokens', 'POST',
                            body=IsA(str), headers=IsA(dict)).\
            AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res200, ''))
        self.mox.ReplayAll()

        self.client.do_request('/resource', 'GET')
        self.assertEqual(self.client.endpoint_url, ENDPOINT_URL)
        self.assertEqual(self.client.auth_token, TOKEN)

    def test_refresh_token(self):
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN
        self.client.endpoint_url = ENDPOINT_URL

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200
        res401 = self.mox.CreateMock(httplib2.Response)
        res401.status = 401

        # If a token is expired, quantum server retruns 401
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res401, ''))
        self.client.request(AUTH_URL + '/tokens', 'POST',
                            body=IsA(str), headers=IsA(dict)).\
            AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)).\
            AndReturn((res200, ''))
        self.mox.ReplayAll()
        self.client.do_request('/resource', 'GET')

    def test_get_endpoint_url(self):
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200

        self.client.request(StrContains(AUTH_URL +
                                        '/tokens/%s/endpoints' % TOKEN), 'GET',
                            headers=IsA(dict)). \
            AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
        self.client.request(StrContains(ENDPOINT_URL + '/resource'), 'GET',
                            headers=ContainsKeyValue('X-Auth-Token', TOKEN)). \
            AndReturn((res200, ''))
        self.mox.ReplayAll()
        self.client.do_request('/resource', 'GET')

    def test_get_endpoint_url_other(self):
        self.client = HTTPClient(username=USERNAME, tenant_name=TENANT_NAME,
                                 password=PASSWORD, auth_url=AUTH_URL,
                                 region_name=REGION, endpoint_type='otherURL')
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200

        self.client.request(StrContains(AUTH_URL +
                                        '/tokens/%s/endpoints' % TOKEN), 'GET',
                            headers=IsA(dict)). \
            AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
        self.mox.ReplayAll()
        self.assertRaises(exceptions.EndpointTypeNotFound,
                          self.client.do_request,
                          '/resource',
                          'GET')

    def test_get_endpoint_url_failed(self):
        self.mox.StubOutWithMock(self.client, "request")

        self.client.auth_token = TOKEN

        res200 = self.mox.CreateMock(httplib2.Response)
        res200.status = 200
#.........这里部分代码省略.........
开发者ID:,项目名称:,代码行数:101,代码来源:


示例17: Client


#.........这里部分代码省略.........
        return self.delete(self.config_handle_path % (config_handle))
    
    @APIParamsCall
    def show_config_handle(self, config_handle, **_params):
        """
        Fetches information of a certain config_handle
        """
        return self.get(self.config_handle_path % (config_handle), params=_params)
        
    @APIParamsCall
    def update_config_handle(self, config_handle, body=None):
        """
        Updates a config_handle
        """
        return self.put(self.config_handle_path % (config_handle), body=body)
        
    @APIParamsCall
    def generate_slb_config(self, body=None):
        """
        Generate the specified configuration
        """
        return self.post(self.slb_configs_path, body=body)
        
    @APIParamsCall
    def launch_chain(self, launch, **_params):
        """
        Generate the specified configuration
        """       
        return self.get(self.launch_path % (launch), params=_params)

    def __init__(self, **kwargs):
        """ Initialize a new client for the Quantum v2.0 API. """
        super(Client, self).__init__()
        self.httpclient = HTTPClient(**kwargs)
        self.version = '2.0'
        self.format = 'json'
        self.action_prefix = "/v%s" % (self.version)
        self.retries = 0
        self.retry_interval = 1

    def _handle_fault_response(self, status_code, response_body):
        # Create exception with HTTP status code and message
        error_message = response_body
        _logger.debug("Error message: %s", error_message)
        # Add deserialized error message to exception arguments
        try:
            des_error_body = Serializer().deserialize(error_message,
                                                      self.content_type())
        except:
            # If unable to deserialized body it is probably not a
            # Quantum error
            des_error_body = {'message': error_message}
        # Raise the appropriate exception
        exception_handler_v20(status_code, des_error_body)

    def do_request(self, method, action, body=None, headers=None, params=None):
        # Add format and tenant_id
        action += ".%s" % self.format
        action = self.action_prefix + action
        if type(params) is dict and params:
            action += '?' + urllib.urlencode(params, doseq=1)
        if body:
            body = self.serialize(body)
        self.httpclient.content_type = self.content_type()
        resp, replybody = self.httpclient.do_request(action, method, body=body)
        status_code = self.get_status_code(resp)
开发者ID:kumarcv,项目名称:openstack-nf,代码行数:67,代码来源:client.py


示例18: Client


#.........这里部分代码省略.........

    @APIParamsCall
    def create_firewall(self, body=None):
        """
        Creates a new firewall
        """
        return self.post(self.firewalls_path, body=body)

    @APIParamsCall
    def update_firewall(self, firewall, body=None):
        """
        Updates a firewall
        """
        return self.put(self.firewall_path % (firewall), body=body)

    @APIParamsCall
    def delete_firewall(self, firewall):
        """
        Deletes the specified firewall
        """
        return self.delete(self.firewall_path % (firewall))

    @APIParamsCall
    def remove_router_from_l3_agent(self, l3_agent, router_id):
        """
        Remove a router from l3 agent.
        """
        return self.delete((self.agent_path + self.L3_ROUTERS + "/%s") % (
            l3_agent, router_id))

    def __init__(self, **kwargs):
        """Initialize a new client for the Quantum v2.0 API."""
        super(Client, self).__init__()
        self.httpclient = HTTPClient(**kwargs)
        self.version = '2.0'
        self.format = 'json'
        self.action_prefix = "/v%s" % (self.version)
        self.retries = 0
        self.retry_interval = 1

    def _handle_fault_response(self, status_code, response_body):
        # Create exception with HTTP status code and message
        _logger.debug("Error message: %s", response_body)
        # Add deserialized error message to exception arguments
        try:
            des_error_body = self.deserialize(response_body, status_code)
        except Exception:
            # If unable to deserialized body it is probably not a
            # Quantum error
            des_error_body = {'message': response_body}
        # Raise the appropriate exception
        exception_handler_v20(status_code, des_error_body)

    def do_request(self, method, action, body=None, headers=None, params=None):
        # Add format and tenant_id
        action += ".%s" % self.format
        action = self.action_prefix + action
        if type(params) is dict and params:
            params = utils.safe_encode_dict(params)
            action += '?' + urllib.urlencode(params, doseq=1)
        # Ensure client always has correct uri - do not guesstimate anything
        self.httpclient.authenticate_and_fetch_endpoint_url()
        uri_len = len(self.httpclient.endpoint_url) + len(action)
        if uri_len > self.MAX_URI_LEN:
            raise exceptions.RequestURITooLong(
                excess=uri_len - self.MAX_URI_LEN)
开发者ID:,项目名称:,代码行数:67,代码来源:



注:本文中的quantumclient.client.HTTPClient类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python v2_0.find_resourceid_by_name_or_id函数代码示例发布时间:2022-05-26
下一篇:
Python quantumclient.Client类代码示例发布时间:2022-05-26
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap