Python urllib.error 模块，code() 实例源码

def execute_get(self, operation, headers={}):
        url = self._host + operation.get_url()
        headers['Content-Type'] = 'application/json;charset=utf-8'
        request = urllib.request.Request(url, headers=headers)

        try:
            response = self._opener.open(request)
        except urllib.error.HTTPError as error:
            error_body = error.read().decode('utf-8')
            new_error_string = ('HTTP error ' +
                                str(error.code) + ' ' +
                                error.reason + ': ' +
                                error_body)
            raise ConnectionError(new_error_string)

        return response.read().decode('utf-8')

def is_operation_supported(self, operation=None, headers={}):
        url = self._host + '/graph/operations/' + operation.get_operation()
        headers['Content-Type'] = 'application/json;charset=utf-8'

        request = urllib.request.Request(url, headers=headers)

        try:
            response = self._opener.open(request)
        except urllib.error.HTTPError as error:
            error_body = error.read().decode('utf-8')
            new_error_string = ('HTTP error ' +
                                str(error.code) + ' ' +
                                error.reason + ': ' +
                                error_body)
            raise ConnectionError(new_error_string)

        response_text = response.read().decode('utf-8')

        return response_text

def main():
    parser = argparse.ArgumentParser()

    parser.add_argument('-q', '--term', dest='term', default=DEFAULT_TERM,
                        type=str, help='Search term (default: %(default)s)')
    parser.add_argument('-l', '--location', dest='location',
                        default=DEFAULT_LOCATION, type=str,
                        help='Search location (default: %(default)s)')

    input_values = parser.parse_args()

    try:
        query_api(input_values.term, input_values.location)
    except urllib.error.HTTPError as error:
        sys.exit(
                'Encountered HTTP error {0}. Abort '
                'program.'.format(error.code))

def getDetailedBinaryLog(self, module, starttime, stoptime, filename):
        url = self.getDetailed_url % {
            "addr": module, "start": starttime, "stop": stoptime}

        if isinstance(self.urlopener, urllib.request.OpenerDirector):
            url_f = self.urlopener.open(url, timeout=30)
        else:
            url_f = self.urlopener.open(url)

        if url_f.code != 200:
            url_f.close()
            raise j.exceptions.RuntimeError(
                "Server fault code %s" % (url_f.code))
        else:
            with open(filename, 'wb') as fp:
                while True:
                    chunk = url_f.read(65536)
                    if not chunk:
                        break
                    fp.write(chunk)
            url_f.close()

def defaultPrintAnswerOut(code,additional):

def http_error_301(self, req, fp, code, msg, headers):
 self.preProcessingRedirection(req, fp, code, msg, headers)     result = super(SmartRedirectHandler, self).http_error_301(req, fp, code, msg, headers) self.postProcessingRedirection(result)
 return result  def http_error_302(self, req, fp, code, msg, headers): self.preProcessingRedirection(req, fp, code, msg, headers)       result = super(SmartRedirectHandler, self).http_error_302(req, fp, code, msg, headers) self.postProcessingRedirection(result) return result
 def preProcessingRedirection(self, req, fp, code, msg, headers):       location = '' for i in headers._headers: if i[0] == 'Location':             location = i[1].strip()         req.add_header('Host',urlparse(location).netloc)
        printAnswer(code, str(msg) + " " + location)        printHeaders(headers._headers,'Set-Cookie')

def requestC(opener,url, headers, data, method = 'POST'):   [answer, code] = requestB(opener,url, headers, data, method)

def test_includes_code_and_msg(self):
        def raise_error():
            raise UCSM_XML_API_Error('bad', 4224)

        error = self.assertRaises(UCSM_XML_API_Error, raise_error)

        self.assertEqual('bad', error.args[0])
        self.assertEqual(4224, error.code)

def _send_data(self, url):
        if isinstance(self.urlopener, urllib.request.OpenerDirector):
            url_f = self.urlopener.open(url, timeout=30)
        else:
            url_f = self.urlopener.open(url)
        data = url_f.read()
        url_f.close()
        if url_f.code != 200:
            raise j.exceptions.RuntimeError(
                "Server fault code %s message %s" % (url_f.code, data))
        return data

def execute_operation_chain(self, operation_chain, headers={}):
        """
        This method queries Gaffer with the provided operation chain.
        """

        # Construct the full URL path to the Gaffer server
        url = self._host + '/graph/operations/execute'

        if hasattr(operation_chain, "to_json"):
            op_chain_json_obj = operation_chain.to_json()
        else:
            op_chain_json_obj = operation_chain

        # Query Gaffer
        if self._verbose:
            print('\nQuery operations:\n' +
                  json.dumps(op_chain_json_obj, indent=4) + '\n')

        # Convert the query dictionary into JSON and post the query to Gaffer
        json_body = bytes(json.dumps(op_chain_json_obj), 'ascii')
        headers['Content-Type'] = 'application/json;charset=utf-8'

        request = urllib.request.Request(url, headers=headers, data=json_body)

        try:
            response = self._opener.open(request)
        except urllib.error.HTTPError as error:
            error_body = error.read().decode('utf-8')
            new_error_string = ('HTTP error ' +
                                str(error.code) + ' ' +
                                error.reason + ': ' +
                                error_body)
            raise ConnectionError(new_error_string)
        response_text = response.read().decode('utf-8')

        if self._verbose:
            print('Query response: ' + response_text)

        if response_text is not None and response_text is not '':
            result = json.loads(response_text)
        else:
            result = None

        return g.JsonConverter.from_json(result)