python类request()的实例源码

def force_update_db():
    dbdir = '/var/tmp/supotato'
    dbpath = '/var/tmp/supotato/supotato.db'
    if os.path.exists(dbpath):
        os.remove(dbpath)

    if not os.path.exists(dbdir):
        os.mkdir(dbdir)

    import urllib.request
    url = "https://everettjf.github.io/app/supotato/supotato.db"

    print('CocoaPods Index Database not exist , will download from ' + url)
    print('Please wait for a moment (about 5 MB file will be downloaded)')

    # Download the file from `url` and save it locally under `file_name`:
    with urllib.request.urlopen(url) as response, open(dbpath, 'wb') as out_file:
        data = response.read() # a `bytes` object
        out_file.write(data)

    print('Download completed')
    print('Update succeed')

def OnClicked(self):
        r = sr.Recognizer()
        with sr.Microphone() as source:
            speak.say('Hey I am Listening ')
            speak.runAndWait()
            audio = r.listen(source)
        try:
            put=r.recognize_google(audio)
            self.displayText(put)
            self.textBox.insert('1.2',put)
            self.textBox.delete('1.2',tk.END)
            events(self,put)
        except sr.UnknownValueError:
            self.displayText("Could not understand audio")
        except sr.RequestError as e:
            self.displayText("Could not request results; {0}".format(e))

def OnClicked(self):
        r = sr.Recognizer()
        with sr.Microphone() as source:
            speak.say('Hey I am Listening ')
            speak.runAndWait()
            audio = r.listen(source)
        try:
            put=r.recognize_google(audio)

            self.displayText(put)
            self.textBox.insert('1.2',put)
            put=put.lower()
            put = put.strip()
            #put = re.sub(r'[?|$|.|!]', r'', put)
            link=put.split()
            events(self,put,link)
        except sr.UnknownValueError:
            self.displayText("Could not understand audio")
        except sr.RequestError as e:
            self.displayText("Could not request results; {0}".format(e))

def setUp(self):
        # the URLs for now which will have the WSDL files and the XSD file
        #urlparse.urljoin('file:', urllib.pathname2url(os.path.abspath("service.xml")))
        import urllib
        import os
        from urllib.parse import urlparse
        from urllib.request import pathname2url

        query_services_url = urllib.parse.urljoin('file:', pathname2url(os.path.abspath('../wsdl_files/vipuserservices-query-1.7.wsdl')))
        userservices_url = urllib.parse.urljoin('file:', pathname2url(os.path.abspath('../wsdl_files/vipuserservices-auth-1.7.wsdl')))

        # initializing the Suds clients for each url, with the client certificate youll have in the same dir as this file
        query_services_client = Client(query_services_url,
                                       transport=HTTPSClientCertTransport('vip_certificate.crt', 'vip_certificate.crt'))
        user_services_client = Client(userservices_url,
                                      transport=HTTPSClientCertTransport('vip_certificate.crt', 'vip_certificate.crt'))

        self.test_user_services_object = SymantecUserServices(user_services_client)

def test_poll_in_Push(self, mock):
        reply = {"requestId": "ac123", "status": "6040", "statusMessage": "Mobile push request sent",
                 "pushDetail": {"pushCredentialId": "133709001", "pushSent": True},
                 "transactionId": "RealTransactionId",
                 "authContext": {"params": {"Key": "authLevel.level", "Value": 10}}}

        mock.SymantecServices.authenticateUserWithPushThenPolling.return_value = Mock()
        mock.authenticateUserWithPushThenPolling.return_value.hash.return_value = reply

        response = symantec_package.lib.allServices.SymantecServices.authenticateUserWithNothing()
        self.assertTrue(response.hash() != reply)

        response = symantec_package.lib.allServices.SymantecServices.authenticateUserWithPushThenPolling("Parameters Here!")

        self.assertTrue((response.hash()) == reply)

        self.assertTrue(response.hash()["status"] == "6040")
        self.assertTrue(response.hash()['requestId'] == "ac123")
        self.assertTrue(response.hash()['statusMessage'] == "Mobile push request sent")
        self.assertTrue(response.hash()["pushDetail"]['pushCredentialId'] == "133709001")
        self.assertTrue(response.hash()["pushDetail"]['pushSent'] is True)
        self.assertTrue(response.hash()['transactionId'] == "RealTransactionId")
        self.assertTrue(response.hash()['authContext']['params']['Key'] == "authLevel.level")
        self.assertTrue(response.hash()['authContext']['params']['Value'] == 10)
        pass

def setUp(self):
        # the URLs for now which will have the WSDL files and the XSD file
        import urllib
        import os
        from urllib.parse import urlparse
        from urllib.request import pathname2url

        managementservices_url = urllib.parse.urljoin('file:', pathname2url(
            os.path.abspath('../wsdl_files/vipuserservices-mgmt-1.7.wsdl')))
        # managementservices_url = 'http://webdev.cse.msu.edu/~huynhall/vipuserservices-mgmt-1.7.wsdl'

        # initializing the Suds clients for each url, with the client certificate youll have in the same dir as this file
        self.management_client = Client(managementservices_url,
                                        transport=HTTPSClientCertTransport('vip_certificate.crt',
                                                                           'vip_certificate.crt'))

        self.test_management_services_object = SymantecManagementServices(self.management_client)
        pass

def test_mock_create_user(self, mock_managementservices):
        reply = {'requestId': 'create_123', 'status': '0000', 'statusMessage': 'Success'}
        # Configure the mock to return a response with an OK status code. Also, the mock should have
        # a `json()` method that returns a list of todos.
        mock_managementservices.createUser.return_value = Mock()
        mock_managementservices.createUser.return_value.json.return_value = reply

        # Call the service, which will send a request to the server.
        response = symantec_package.lib.managementService.SymantecManagementServices.createUser("create_123",
                                                                                                "new_user3")

        print(response.json())
        # If the request is sent successfully, then I expect a response to be returned.
        self.assertTrue((response.json()) == reply)
        self.assertTrue((response.json()["status"] == "0000"))
        pass

def test_mock_delete_user(self, mock_managementservices):
        reply = {'requestId': 'delete_123', 'status': '0000', 'statusMessage': 'Success'}

        # Configure the mock to return a response with an OK status code. Also, the mock should have
        # a `json()` method that returns a list of todos.
        mock_managementservices.deleteUser.return_value = Mock()
        mock_managementservices.deleteUser.return_value.json.return_value = reply

        # Call the service, which will send a request to the server.
        response = symantec_package.lib.managementService.SymantecManagementServices.deleteUser("delete_123",
                                                                                                "new_user3")

        print(response.json())
        # If the request is sent successfully, then I expect a response to be returned.
        self.assertTrue((response.json()) == reply)
        self.assertTrue((response.json()["status"] == "0000"))
        pass

def test_mock_add_STANDARDOTP_credential(self, mock_managementservices):
        reply = {'statusMessage': "Success", 'requestId': 'add_otp_cred', 'status': '0000'}

        # Configure the mock to return a response with an OK status code. Also, the mock should have
        # a `json()` method that returns a list of todos.
        mock_managementservices.addCredentialOtp.return_value = Mock()
        mock_managementservices.addCredentialOtp.return_value.json.return_value = reply

        response = symantec_package.lib.managementService.SymantecManagementServices.addCredentialOtp("add_otp_cred", "new_user3",
                                                                               "", "STANDARD_OTP", \
                                                                               "678066")  # change with what's on your device
        print(response.json())
        # If the request is sent successfully, then I expect a response to be returned.
        self.assertTrue((response.json()) == reply)
        self.assertTrue((response.json()["status"] == "0000"))
        pass

def test_mock_update_STANDARDOTP_credential(self, mock_managementservices):
        reply = {'statusMessage': 'Success', 'requestId': 'update_123', 'status': '0000'}

        # Configure the mock to return a response with an OK status code. Also, the mock should have
        # a `json()` method that returns a list of todos.
        mock_managementservices.updateCredential.return_value = Mock()
        mock_managementservices.updateCredential.return_value.json.return_value = reply

        response = symantec_package.lib.managementService.SymantecManagementServices.updateCredential("update_123", "gabe_phone",
                                                                                                      "", "STANDARD_OTP",
                                                                                                      "My personal cell phone")
        print(response.json())
        # If the request is sent successfully, then I expect a response to be returned.
        self.assertTrue((response.json()) == reply)
        self.assertTrue((response.json()["status"] == "0000"))
        pass

def test_mock_setTemporaryPasswordSMSDelivery(self, mock_managementservices):
        reply = {'status': '0000', 'requestId': 'setTempPWD', 'statusMessage': 'Success', 'temporaryPassword': '998241'}

        # Configure the mock to return a response with an OK status code. Also, the mock should have
        # a `json()` method that returns a list of todos.
        mock_managementservices.setTemporaryPasswordSMSDelivery.return_value = Mock()
        mock_managementservices.setTemporaryPasswordSMSDelivery.return_value.json.return_value = reply

        response = symantec_package.lib.managementService.SymantecManagementServices.setTemporaryPasswordSMSDelivery("setTempPWD",
                                                                                                      "gabe_phone",
                                                                                                      "12313608781",
                                                                                                      "17879481605")
        print(response.json())
        # If the request is sent successfully, then I expect a response to be returned.
        self.assertTrue((response.json()) == reply)
        self.assertTrue((response.json()["status"] == "0000"))
        pass

def sendMessage(self, data, method, url):
        conn = http.client.HTTPConnection(Parameters["Address"] + ":1400")
        headers = {"Content-Type": 'text/xml; charset="utf-8"', "SOAPACTION": method}
        conn.request("POST", url, data, headers)
        response = conn.getresponse()
        conn.close()

        if response.status == 200:
            data = response.read().decode("utf-8")
            LogMessage(str(data))
            self.parseMessage(data)
        else:
            Domoticz.Error("Unexpected response status received in function sendMessage (" + str(response.status) + ", " + str(response.reason) + "). \
                            The following command is sent: " + str(method) + ", " + str(url))
        return

    # Process message from Sonos

def getRecognitionResult(self, apiName, seqValue):
        query = self.getBasicQueryString(apiName, seqValue) + "&stop=1"

        '''Request speech recognition service by HTTP GET'''
        url = str(self.apiBaseUrl) + "?" + str(query)
        req = urllib.request.Request(url,headers = {'Cookie': self.cookies})
        with urllib.request.urlopen(req) as f:
            getResponse = f.read().decode()

        '''Now you can check the status here.'''
        print("Sending 'GET' request to URL : " + self.apiBaseUrl)
        print("get parameters : " + str(query))
        print("Response Code : " + str(f.getcode()))

        '''Get the response'''
        return str(getResponse)

def load_url(url, timeout):
    # Build URL query to email signup page
    urlquery = "http://" + url + "/m-users-a-email_list-job-add-email-" + targetEmail + "-source-2.htm"
    print_out(Style.BRIGHT + Fore.WHITE + "Sending request to: " + url)
    # Build the request
    req = urllib.request.Request(
        urlquery, 
        data=None, 
        headers={
            'User-Agent': random.choice(useragents),
            'Host': url
        }
    )
    # Send
    try:
        f = urllib.request.urlopen(req)
        print_out(Style.BRIGHT + Fore.GREEN + "Successfully sent!")
        f.close()
    except urllib.error.URLError as e:
        print_out(Style.BRIGHT + Fore.RED + e.reason)

def retry(max_retries):
    """
    Retry a function `max_retries` times.
    taken from https://stackoverflow.com/questions/23892210/python-catch-timeout-and-repeat-request.
    """
    def retry(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            num_retries = 0
            while num_retries <= max_retries:
                try:
                    ret = func(*args, **kwargs)
                    break
                except HTTPError:
                    if num_retries == max_retries:
                        raise
                    num_retries += 1
                    time.sleep(1)
            return ret
        return wrapper
    return retry

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 paradasBiziCercanas(bot,update,longitud,latitud,estado,numposte):
    DISTANCIA = '350'#Distancia en metros desde la posición enviada, lo ponemos como string para evitarnos conversiones luego. Por contrato la más cercana tiene que estar como mucho a 300 metros, así que al poner 350 nos aseguramos de que haya al menos otra
    url='http://www.zaragoza.es/api/recurso/urbanismo-infraestructuras/estacion-bicicleta.json?rf=html&results_only=false&srsname=wgs84&point='+longitud+','+latitud+'&distance='+DISTANCIA
    try:
        h = urllib.request.urlopen(url)
    except Exception as e:
        bot.sendMessage(chat_id=update.message.chat_id, text='??<b>Error</b>??\nImposible contactar con el servicio del Ayuntamiento.', parse_mode='HTML')
    #BIZI
    jsonleidobizi = json.loads(str(h.read().decode('utf-8')))
    h.close()
    nElementosbizi = jsonleidobizi["totalCount"]
    textobizi = ''
    if nElementosbizi==0:
        textobizi='No hay estaciones BiZi a '+DISTANCIA+' metros de la ubicación\n\n'
    else:
        for i in range(nElementosbizi):
            if(jsonleidobizi["result"][i]["id"]!=numposte):#no enseñar la parada desde la que se invoca
                textobizi = textobizi + '/bizi '+ jsonleidobizi["result"][i]["id"] + '\n' + jsonleidobizi["result"][i]["title"] + '\n\n'
    bot.sendMessage(chat_id=update.message.chat_id, text='<b>'+estado+', quizás te interesen otras estaciones cercanas (a &lt'+DISTANCIA+'m):</b>\n'+textobizi, parse_mode='HTML', disable_web_page_preview=True)

def get_api_raw(self, url):
        request = urllib.request.Request(self._api_url + url)
        try:
            result = urllib.request.urlopen(request)
        except urllib.error.HTTPError as e:
            self._error = "HTTP error"
            self._error_msg = str(e.code)
            self._update_ready = None 
        except urllib.error.URLError as e:
            self._error = "URL error, check internet connection"
            self._error_msg = str(e.reason)
            self._update_ready = None 
            return None
        else:
            result_string = result.read()
            result.close()
            return result_string.decode()
        # if we didn't get here, return or raise something else


    # result of all api calls, decoded into json format

def get_api_raw(self, url):
        request = urllib.request.Request(self._api_url + url)
        try:
            result = urllib.request.urlopen(request)
        except urllib.error.HTTPError as e:
            self._error = "HTTP error"
            if str(e.code) == '404':
                self._error_msg = "404 - repository not found, verify register settings"
            else:
                self._error_msg = "Response: "+str(e.code)
            self._update_ready = None 
        except urllib.error.URLError as e:
            self._error = "URL error, check internet connection"
            self._error_msg = str(e.reason)
            self._update_ready = None 
            return None
        else:
            result_string = result.read()
            result.close()
            return result_string.decode()
        # if we didn't get here, return or raise something else


    # result of all api calls, decoded into json format