Python requests.exceptions 模块，MissingSchema() 实例源码

def saveFile(self, url, page, idx):
        user_define_name = self.now_date() + '_p_' + str(page) + '_' + string.zfill(idx, 2)  # ??2?
        file_ext = self.file_extension(url)  # ???
        save_file_name = user_define_name + "_" + file_ext

        # ???????open??
        # urllib.urlretrieve(item[0], self.save_path + save_file_name)
        # ????
        url = self.CheckUrlValidate(url)
        try:
            pic = requests.get(url, timeout=10)
            f = open(self.store_dir + os.sep + save_file_name, 'wb')
            f.write(pic.content)
            f.close()
            print '\ndone save file ' + save_file_name
        except ReadTimeout:
              print 'save file %s failed. cause by timeout(10)' %(save_file_name)
        except MissingSchema:
            print 'invalid url %s' %(url)
        except Exception, e:
            print e


    #??url????http:??

def test_connerror_missing_url_schema(self):
        # GIVEN server url with missing schema
        self.config['server'] = '127.0.0.1'

        # WHEN dpm publish is invoked
        try:
            result = self.invoke(cli, ['publish', ])
        except Exception as e:
            result = e

        # THEN MissingSchema should be raised
        assert isinstance(result, MissingSchema)
        # AND it should say that schema is invalid
        assert "Invalid URL '127.0.0.1/api/auth/token': No schema supplied. Perhaps you meant http://127.0.0.1/api/auth/token?" in str(result)

def test_requests_instrumentation_malformed_none(elasticapm_client):
    elasticapm_client.begin_transaction("transaction.test")
    with capture_span("test_request", "test"):
        with pytest.raises(MissingSchema):
            requests.get(None)

def test_requests_instrumentation_malformed_schema(elasticapm_client):
    elasticapm_client.begin_transaction("transaction.test")
    with capture_span("test_request", "test"):
        with pytest.raises(MissingSchema):
            requests.get('')

def gbf(bot, message):
    result = GBF_PATTERN.findall(message.text)
    for index, match in enumerate(result):
        if index > 10:
            return False
        url = match[0]
        if url.split('.')[-1] in ['png', 'jpg', 'jpeg', 'gif']:
            return False
        if url not in match_urls:
            try:
                resp = requests.get(url)
            except MissingSchema:
                resp = requests.get('http://' + url)
            if 'granblue' in resp.url:
                match_urls.append(url)
                reply(bot, message, '?????????')
                return True
        else:
            reply(bot, message, '?????????')
            return True

def check_sites(self):
        errors = []

        try:
            errors = [site.create_slack_message() for site in self.sites
                      if not site.check_status_code()]
        except (ConnectionError, MissingSchema, Timeout) as e:
            self.slack.post_message(unicode(e.message))
            self.logger.error(unicode(e.message))

        for error in errors:
            self.slack.post_message(error)
            self.logger.error(error)

        return len(errors)

def test_get_status_code_with_bad_url(self):
        # No prefix or suffixes
        bad_url = "url"
        expected_status_code = 200

        monitor = MonitorSite(url=bad_url,
                              expected_status_code=expected_status_code)

        self.assertRaises(MissingSchema, monitor.get_status_code)

        bad_url = "url.com"
        expected_status_code = 200

        monitor = MonitorSite(url=bad_url,
                              expected_status_code=expected_status_code)

        self.assertRaises(MissingSchema, monitor.get_status_code)

def md5_sum_from_url(url):
    try:
        r = requests.get(url, stream=True)
    except InvalidSchema:
        return None
    except MissingSchema:
        return None
    chunk_counter = 0
    hash_store = hashlib.md5()
    for chunk in r.iter_content(chunk_size=1024):
        if chunk:  # filter out keep-alive new chunks
            hash_store.update(chunk)
            chunk_counter += 1
            # It is not possible to send greater than 50 MB via Telegram
            if chunk_counter > 50 * 1024:
                return None
    return hash_store.hexdigest()

def test_load_invalid_remote_template_invalid(self):
        """
        Test if it raises an error with an invalid
        schema URL.
        """
        self.assertRaises(MissingSchema,
                          load_remote_schema,
                          "some.incorrect/..url")

def authenticate(func):
        """Decorator that authenticates credentials.

        :type func: callable
        :param func: A method to execute if authorization passes.

        :return: The return value of `func` if authorization passes, or
            None if authorization fails.
        """
        def auth_wrapper(self, *args, **kwargs):
            self.config.authenticate()
            self.config.save_config()
            if self.config.check_auth():
                try:
                    return func(self, *args, **kwargs)
                except SSLError:
                    click.secho(('SSL cert verification failed.\n  Try running '
                                 'gh configure --enterprise\n  and type '
                                 "'n' when asked whether to verify SSL certs."),
                                fg=self.config.clr_error)
                except MissingSchema:
                    click.secho('Invalid GitHub Enterprise url',
                                fg=self.config.clr_error)
                except AuthenticationFailed:
                    self.config.print_auth_error()
        return auth_wrapper

def is_uri_reachable(uri: str) -> bool:
        try:
            return requests.head(uri).status_code == requests.codes.ok
        except MissingSchema:
            return False

    ##
    # Identifies the kind of the source and initializes it
    #
    # @param args   The arguments passed to the application

def load_synonyms(path):
  try:
    r = requests.get(path)
    content = r.text
  except (MissingSchema, InvalidSchema, InvalidURL):
    try:
      with open(path, encoding='utf-8') as fp:
        content = fp.read()
    except (OSError, IOError):
      raise TypeError('Invalid Path: "{0}". Ensure it is either a URL or Correct FS Path.'.format(path))

  return SynParser.get_mapping(content)

def test_setup_missing_schema(self):
        """Test setup with resource missing schema."""
        with self.assertRaises(MissingSchema):
            rest.setup_platform(self.hass, {
                'platform': 'rest',
                'resource': 'localhost',
                'method': 'GET'
            }, None)