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

Python treq.get函数代码示例

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

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



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

示例1: main

def main(reactor):
    print('List of tuples')
    resp = yield treq.get('http://httpbin.org/get',
                          params=[('foo', 'bar'), ('baz', 'bax')])
    content = yield resp.text()
    print(content)

    print('Single value dictionary')
    resp = yield treq.get('http://httpbin.org/get',
                          params={'foo': 'bar', 'baz': 'bax'})
    content = yield resp.text()
    print(content)

    print('Multi value dictionary')
    resp = yield treq.get('http://httpbin.org/get',
                          params={'foo': ['bar', 'baz', 'bax']})
    content = yield resp.text()
    print(content)

    print('Mixed value dictionary')
    resp = yield treq.get('http://httpbin.org/get',
                          params={'foo': ['bar', 'baz'], 'bax': 'quux'})
    content = yield resp.text()
    print(content)

    print('Preserved query parameters')
    resp = yield treq.get('http://httpbin.org/get?foo=bar',
                          params={'baz': 'bax'})
    content = yield resp.text()
    print(content)
开发者ID:JunaidLoonat,项目名称:treq,代码行数:30,代码来源:query_params.py


示例2: test_has_header

    def test_has_header(self):
        class Api(object):
            app = Klein()

            @app.route('/')
            @validate(has_header('X-Foo'))
            def route(self, req):
                pass

        srv = yield ToyServer.from_test(self, Api().app)
        resp = yield treq.get(srv.url, persistent=False)

        self.assertEqual(json.loads((yield resp.content())), {
            'errors': [{
                'type': 'header_missing',
                'message': "Header 'X-Foo' is missing"
            }]
        })

        resp = yield treq.get(
            srv.url,
            headers={'X-Foo': ['bar']},
            persistent=False)

        self.assertEqual(resp.code, http.OK)
开发者ID:praekelt,项目名称:vumi-http-retry-api,代码行数:25,代码来源:test_validate.py


示例3: test_body_schema

    def test_body_schema(self):
        class Api(object):
            app = Klein()

            @app.route('/')
            @json_body
            @validate(body_schema({'properties': {'foo': {'type': 'string'}}}))
            def route(self, req, body):
                pass

        srv = yield ToyServer.from_test(self, Api().app)
        resp = yield treq.get(
            srv.url,
            persistent=False,
            data=json.dumps({'foo': 23}))

        content = yield resp.json()
        self.assertEqual(content['result'], {
            'errors': [{
                'type': 'invalid_body',
                'message': "23 is not of type 'string'",
                'schema_path': ['properties', 'foo', 'type'],
            }]
        })
        self.assertEqual(content['status'], 400)
        self.assertEqual(content['code'], 'Bad Request')
        self.assertEqual(content['description'], 'api usage error')

        resp = yield treq.get(
            srv.url,
            persistent=False,
            data=json.dumps({'foo': 'bar'}))

        self.assertEqual(resp.code, http.OK)
开发者ID:praekelt,项目名称:junebug,代码行数:34,代码来源:test_validate.py


示例4: test_body_schema

    def test_body_schema(self):
        class Api(object):
            app = Klein()

            @app.route('/')
            @json_body
            @validate(body_schema({'properties': {'foo': {'type': 'string'}}}))
            def route(self, req, body):
                pass

        srv = yield ToyServer.from_test(self, Api().app)
        resp = yield treq.get(
            srv.url,
            persistent=False,
            data=json.dumps({'foo': 23}))

        self.assertEqual(json.loads((yield resp.content())), {
            'errors': [{
                'type': 'invalid_body',
                'message': "23 is not of type 'string'"
            }]
        })

        resp = yield treq.get(
            srv.url,
            persistent=False,
            data=json.dumps({'foo': 'bar'}))

        self.assertEqual(resp.code, http.OK)
开发者ID:praekelt,项目名称:vumi-http-retry-api,代码行数:29,代码来源:test_validate.py


示例5: process_item

    def process_item(self, item, spider):
        try:
            if type(item) is UrlItem:
                # Call Pagespeed API
                url = item.get('url', '')
                params={'url' : url, 'key' : self.api_key }
                #logging.debug('PageSpeedPipeline:' + self.PAGESPEED_URL + str(params))

                treq.get(self.PAGESPEED_URL, params=params).addCallback(self.api_done)
                #r = requests.get(self.PAGESPEED_URL, params=params)

                '''
                response = yield treq.get(self.PAGESPEED_URL, params=params)
                code = yield treq.code(response)

                if (code is 200):
                    result = yield treq.text(response)

                    with self.connection.cursor() as cursor:
                        # Store result in DB
                        cursor.execute("""INSERT INTO pagespeed (url, result) VALUES(%s, %s)""", ( url, result ) )
                        self.connection.commit()
                '''
                            
        except:
            logging.exception("PageSpeedPipeline Error %s", item.get('url', ''))
            
        return item
开发者ID:natoinet,项目名称:seoscraper,代码行数:28,代码来源:pipelines.py


示例6: test_custom_agent

 def test_custom_agent(self):
     """
     A custom Agent is used if specified.
     """
     custom_agent = mock.Mock()
     treq.get('https://www.example.org/', agent=custom_agent)
     self.HTTPClient.assert_called_once_with(custom_agent)
开发者ID:JayH5,项目名称:treq,代码行数:7,代码来源:test_api.py


示例7: main

def main(reactor):
    tor = yield txtorcon.connect(
        reactor,
        UNIXClientEndpoint(reactor, "/var/run/tor/control")
    )

    print("Connected to Tor version {}".format(tor.version))

    url = 'https://www.torproject.org:443'
    print("Downloading {}".format(url))
    resp = yield treq.get(url, agent=tor.web_agent())

    print("   {} bytes".format(resp.length))
    data = yield resp.text()
    print("Got {} bytes:\n{}\n[...]{}".format(
        len(data),
        data[:120],
        data[-120:],
    ))

    print("Creating a circuit")
    state = yield tor.create_state()
    circ = yield state.build_circuit()
    yield circ.when_built()
    print("  path: {}".format(" -> ".join([r.ip for r in circ.path])))

    print("Downloading meejah's public key via above circuit...")
    config = yield tor.get_config()
    resp = yield treq.get(
        'https://meejah.ca/meejah.asc',
        agent=circ.web_agent(reactor, config.socks_endpoint(reactor)),
    )
    data = yield resp.text()
    print(data)
开发者ID:felipedau,项目名称:txtorcon,代码行数:34,代码来源:readme.py


示例8: start_streaming

def start_streaming(host, user):
    print "Initial Sync for testuser_%d" % (user,)

    r = yield treq.get(
        "%s/_matrix/client/api/v1/events" % (host,),
        params={
            "access_token": "token_%d" % (user,),
            "timeout": "0",
        },
    )

    initialSync = yield r.json()

    assert r.code == 200, "InitialSync %d. %r" % (r.code, initialSync,)

    from_token = initialSync["end"]

    yield sleep(5 * random.random())  # This is to even requests out.

    print "Starting event stream for testuser_%d" % (user,)

    while True:
        try:
            start = get_time()
            r, i = yield defer.DeferredList(
                [
                    treq.get(
                        "%s/_matrix/client/api/v1/events" % (host,),
                        params={
                            "access_token": "token_%d" % (user,),
                            "from": from_token,
                            "timeout": "30000",
                        },
                    ),
                    sleep(45),
                ],
                fireOnOneCallback=True,
            )
            end = get_time()

            if i == 1:
                # We timedout.
                print "Cancelled after %ds, retrying" % ((end-start)/1000,)
                continue

            stream = yield r.json()

            assert r.code == 200, "Event stream %d. %r" % (r.code, stream,)

            if end - start > 35000:
                print "Took more than %ds to get response" % ((end-start)/1000,)

            from_token = stream["end"]
        except Exception as e:
            print "Failed to do event stream for %d: %s" % (user, e)
            yield sleep(5)
开发者ID:matrix-org,项目名称:vagrant-synapse-test,代码行数:56,代码来源:event_streams.py


示例9: test_api_dump_invalid_querystring

    def test_api_dump_invalid_querystring(self):
        yield ShortenerTables(self.account, self.conn).create_tables()

        url = "http://en.wikipedia.org/wiki/Cthulhu"
        yield self.service.shorten_url(url, "test-user")
        yield treq.get(self.make_url("/qr0"), allow_redirects=False, pool=self.pool)

        resp = yield treq.get(self.make_url("/api/handler/dump"), allow_redirects=False, pool=self.pool)

        self.assertEqual(resp.code, 400)
开发者ID:praekelt,项目名称:url-shortening-service,代码行数:10,代码来源:test_handlers.py


示例10: test_cached_pool

    def test_cached_pool(self):
        pool = self.HTTPConnectionPool.return_value

        treq.get('http://test.com')

        self.HTTPConnectionPool.return_value = mock.Mock()

        treq.get('http://test.com')

        self.Agent.assert_called_with(mock.ANY, pool=pool)
开发者ID:FxIII,项目名称:treq,代码行数:10,代码来源:test_api.py


示例11: lookupUri

    def lookupUri(self, message):
        if not message.target in self.config:
            return

        for uri in re.findall("https?://[^\s]+", message.message):
            treq.get(uri, unbuffered = True).addCallbacks(
                callback = self._gotResponse,
                errback = self._errorResult,
                callbackArgs = (message, uri),
                errbackArgs = (message, uri)
            )
开发者ID:DASPRiD,项目名称:DASBiT,代码行数:11,代码来源:urilookup.py


示例12: lookupCharacters

def lookupCharacters(member, characters):
  response = yield treq.get("http://www.bungie.net/Platform/User/GetBungieAccount/" + member["membershipId"].encode("UTF-8") + "/254/")
  data = yield treq.json_content(response)
  for account in data["Response"]["destinyAccounts"]:
    if account["userInfo"]["membershipType"] == 1:
      platform = "xbox"
    elif account["userInfo"]["membershipType"] == 2:
      platform = "playstation"
    else:
      continue

    for character in account["characters"]:
      character_data = {
        "bungieId": member["membershipId"],
        "accountId": account["userInfo"]["membershipId"],
        "characterId": character["characterId"],
        "name": account["userInfo"]["displayName"],
        "race": character["race"]["raceName"],
        "gender": character["gender"]["genderName"],
        "class": character["characterClass"]["className"],
        "level": character["level"],
        "levelString": "{:,d}".format(character["level"]),
        "icon": "http://bungie.net" + character["emblemPath"],
        "background": "http://bungie.net" + character["backgroundPath"],

        # Default values for extra data (in case it fails)
        "light": 0,
        "lightString": "{:,d}".format(0),
        "grimoire": 0,
        "grimoireString": "{:,d}".format(0),
        "minutesPlayed": 0,
        "minutesPlayedString": "{:,d}".format(0),
        "lastSeen": "",
        "lastSeenString": ""
      }
      character_data["style"] = 'background: url("' + character_data["background"] + '")'

      try:
        response = yield treq.get("http://www.bungie.net/Platform/Destiny/{!s}/Account/{!s}/Character/{!s}/".format(account["userInfo"]["membershipType"], account["userInfo"]["membershipId"], character["characterId"]))
        extra_data = yield treq.json_content(response)
        extra_data = extra_data["Response"]["data"]["characterBase"]
        character_data["light"] = extra_data["stats"]["STAT_LIGHT"]["value"]
        character_data["lightString"] = "{:,d}".format(character_data["light"])
        character_data["grimoire"] = extra_data["grimoireScore"]
        character_data["grimoireString"] = "{:,d}".format(character_data["grimoire"])
        character_data["minutesPlayed"] = int(extra_data["minutesPlayedTotal"])
        character_data["minutesPlayedString"] = "{:,d}".format(character_data["minutesPlayed"])
        character_data["lastSeen"] = extra_data["dateLastPlayed"]
        character_data["lastSeenString"] = datetime.datetime.strptime(extra_data["dateLastPlayed"], "%Y-%m-%dT%H:%M:%SZ").strftime("%B %d, %I:%M%p")
      except:
        pass

      characters[platform].append(character_data)
开发者ID:glacials,项目名称:destiny,代码行数:53,代码来源:app.py


示例13: check

    def check(self):
        # Hax
        try:
            irc = self.master.modules["irc"]
            words = yield self.config.get("words", [])

            response = yield treq.get("http://a.4cdn.org/a/threads.json")
            threads = yield treq.json_content(response)

            nthreads, check = {}, []
            for o in threads:
                for t in o["threads"]:
                    nthreads[t["no"]] = t["last_modified"]
                    if t["no"] not in self.threads:
                        self.threads[t["no"]] = 0
                    if t["last_modified"] > self.threads[t["no"]]:
                        check.append(t["no"])

            if not self.initial:
                threads = []
                for t in check:
                    response = yield treq.get("http://a.4cdn.org/a/res/{:d}.json".format(t))
                    if response.code == 200:
                        data = yield treq.json_content(response)
                        found, posts = set(), []
                        for p in data["posts"]:
                            if p["time"] > self.threads[t] and "com" in p:
                                f = set(filter(lambda x: re.search(r"\b" + x.lower() + r"\b", p["com"].lower()), words))
                                if f:
                                    found.update(f)
                                    posts.append((p["no"], p["com"]))
                        if found and posts:
                            threads.append((t, found, posts))
                if len(threads) < 5:
                    for t, found, posts in threads:
                        p, also = posts[0], (u"(also {} in same thread)".format(", ".join([str(x[0]) for x in posts[1:]])) if posts[1:] else u"")
                        url = "https://archive.foolz.us/a/thread/{:d}/#{:d}".format(t, p[0])
                        #response = yield treq.post("https://www.googleapis.com/urlshortener/v1/url?key="+API_KEY, json.dumps({"longUrl": url}), headers={'Content-Type': ['application/json']})
                        #data = yield treq.json_content(response)
                        #url = data["id"]
                        excerpt = p[1].replace("<br>", "\n")
                        excerpt = BeautifulSoup(excerpt).get_text()
                        excerpt = re.sub("\s+", " ", excerpt)
                        excerpt = excerpt if len(excerpt) <= 100 else excerpt[:97]+"..."
                        irc.msg(u"#commie-subs", u"\u00039>{} mentioned on /a/ at {} [{}] {}".format(u", ".join(found), url, excerpt, also))
                else:
                    t, found = [str(x[0]) for x in threads], reduce(lambda a,b: a|b, [x[1] for x in threads])
                    irc.msg(u"#commie-subs", u"\u00039>{} flooded /a/ in threads {}".format(u", ".join(found), u", ".join(t)))

            self.threads = nthreads
            self.initial = False
        except:
            self.err(u"4chan module broke horribly :(")
开发者ID:Fugiman,项目名称:Servrhe,代码行数:53,代码来源:4chan.py


示例14: geocode

    def geocode(self, address):
        """
        This method makes a call to Google's geocoding API. You shouldn't
        call this more than 5 times per second
        """

        # The url for this API
        #endpoint = 'https://maps.googleapis.com/maps/api/geocode/json'
        endpoint = 'http://web:9312/maps/api/geocode/json'

        # Do the call
        parms = [('address', address), ('sensor', 'false')]
        response = yield treq.get(endpoint, params=parms)

        # Decode the response as json
        content = yield response.json()

        # If the status isn't ok, return it as a string
        if content['status'] != 'OK':
            raise Exception('Unexpected status="%s" for address="%s"' %
                            (content['status'], address))

        # Extract the address and geo-point and set item's fields
        geo = content['results'][0]["geometry"]["location"]

        # Return the final value
        defer.returnValue({"lat": geo["lat"], "lon": geo["lng"]})
开发者ID:Fighting-Toghter,项目名称:scrapybook,代码行数:27,代码来源:geo.py


示例15: lookupMembers

def lookupMembers(id):
  page = 1
  hasMore = True if id else False
  members = []
  characters = {
    "playstation": [],
    "xbox": []
  }
  deferreds = []

  while hasMore:
    # No idea what is different between V1, V2, and V3...
    response = yield treq.get("http://www.bungie.net/Platform/Group/{!s}/MembersV3/".format(id), params={
      "itemsPerPage": 50,
      "currentPage": page
    })
    data = yield treq.json_content(response)
    page += 1
    hasMore = data["Response"]["hasMore"]
    members.extend(data["Response"]["results"])

  # Load character data in parallel
  for member in members:
    deferreds.append(lookupCharacters(member, characters))
  yield DeferredList(deferreds, fireOnOneErrback=True, consumeErrors=True)

  for platform_characters in characters.values():
    platform_characters.sort(key=lambda c: (c["level"], c["light"]), reverse=True)

  returnValue(characters)
开发者ID:Fugiman,项目名称:destiny,代码行数:30,代码来源:app.py


示例16: _send_request

 def _send_request(self):
     if not self.queue:
         if self.request_loop.running:
             self.request_loop.stop()
         return
     now = time.time() - 1  # 1 second buffer
     if (self.rate_remaining < 1+1 and self.rate_reset > now or 
             DiscordRestApiLoop.global_wait > now):
         self.log.warn("Rate limited: {}".format(self.channel_id))
         return
     payload = self.queue.pop()
     method = payload['method']
     url = payload['url']
     content = payload['content']
    # url = '{}/channels/{}/messages'.format(HOST, self.channel_id)
     content = json.dumps({"content": content})
     self.log.debug('at _send_request: {} url {}'.format(self.channel_id,
                                                         url))
     if method == 'post':
         d = treq.post(url, content, headers=HEADERS)
     elif method == 'patch':
         d = treq.patch(url, content, headers=HEADERS)
     elif method == 'delete':
         d = treq.delete(url, headers=HEADERS)
     elif method == 'get':
         d = treq.get(url, headers=HEADERS)
     d.addCallback(self.update_rate_limits)
     if not self.queue:
         self.request_loop.stop()
开发者ID:d-dd,项目名称:Yukari,代码行数:29,代码来源:dcrestclient.py


示例17: get_details

def get_details(request):
	if not request.args.get('url',[0]):
		# abort(400)
		pass
	else:	
		add_url = "/trustpass_profile.html?certification_type=intl_av"
		try:
			url = request.args.get('url',[0])[0].strip()
			print url
			response = yield treq.get(url.split(".html")[0] + add_url)
		except Exception ,e:
			print e
			log.msg(e,logLevel=logging.DEBUG)
			# response.close()
			# abort(404)
			pass

		# print response2.status_code
		try:
			content = yield response.content()
			soup = BeautifulSoup(content,"html.parser")
			details = soup.find_all("table",{"class":"table"})
			basic_details = details[0].find_all("tr")
			personl_details = details[1].find_all("tr")
			outdict = {}
			for idx,lines in enumerate(basic_details):
				outdict[clean_text(lines,'th')] = clean_text(lines,'td')
				# print clean_text(lines,'td')
			for lines in personl_details:
				outdict[clean_text(lines,'th')] = clean_text(lines,'td')
			returnValue(json.dumps(outdict))
		except KeyError:
				# abort(400)
				pass
开发者ID:faureka,项目名称:scrapy_projects,代码行数:34,代码来源:alibaba_api.py


示例18: check_status

def check_status(request):
	if not request.args.get("cname"):
		pass
		# abort(400)
	# elif not re.match(request.args.get("cname")):	
	else:
		get_url = "http://india.alibaba.com/supplier_list.htm?bizType=&SearchText="
		search_query = request.args.get("cname",[0])[0].strip().replace('.','').replace(',','').upper()
		if re.search(comp_name_regex,search_query):
			search_query = search_query.replace(" ","%20")
			# print search_query
			try:
				response = yield treq.get(get_url+search_query)
				# print response1.status_code
			except Exception,e:
				print e
				log.msg(e,logLevel=logging.DEBUG)
				# response.close()
				abort(404)
			content = yield response.content()	
			soup = BeautifulSoup(content,"html.parser")
			# print soup.find('title').get_text()
			name_comps = soup.find_all("a",{"class":"dot-company"})
			name_url_list = [] 
			for names in name_comps:
				name_url_dict = {}
				name_url_dict['Name'] = names.get("title")
				name_url_dict['href'] = names.get("href")
				name_url_list.append(name_url_dict)
			# response.close()
		returnValue(json.dumps(name_url_list))
开发者ID:faureka,项目名称:scrapy_projects,代码行数:31,代码来源:alibaba_api.py


示例19: joke

def joke(bot):
    """Chuck Norris jokes from http://icndb.com"""
    url = "http://api.icndb.com/jokes/random/1"

    @defer.inlineCallbacks
    def _tell_joke(response, channel, name=None):
        data = yield response.json()
        cnjoke = data['value'][0]['joke']
        cnjoke = cnjoke.replace("&quot;", "\"")
        if name:
            cnjoke = cnjoke.replace("Chuck Norris", name)
        bot.msg(channel, str(cnjoke), length=510)

    while True:
        args, sender, senderhost, channel = yield
        get(url).addCallback(_tell_joke, channel, " ".join(args))
开发者ID:DefaultUser,项目名称:PyTIBot,代码行数:16,代码来源:fun.py


示例20: lookup_geocode

def lookup_geocode(location, api_key=''):
    """
    Determine the lat/long coordinates for a location name.

    :param location: location name
    :type location: str
    :type api_key: str

    :return: a dict with keys 'lat', 'lng', 'loc' that contain
     the lat/long coordinates and placename for the location.
    :rtype: defer.Deferred
    """
    try:
        res = yield treq.get("http://maps.googleapis.com/maps/api/geocode/json",
                             params={'address': location, 'sensor': 'false', 'key': api_key})
        if res and res.code == 200:
            data = yield treq.json_content(res)
            if data['status'] == 'OK':
                # API returned at least one geocode
                ret = data['results'][0]['geometry']['location']
                ret['loc'] = data['results'][0]['formatted_address']
                defer.returnValue(ret)
            else:
                log.warn("Bad status response from Google Geocode API: %s" % (data['status']))
        elif res is not None:
            log.warn("Bad HTTP status from Google Geocode API: %s" % (res.code))
    except TypeError:
        log.warn("Bad location passed to lookup_geocode", exc_info=True)
开发者ID:goudacoin,项目名称:twobitbot,代码行数:28,代码来源:googleapis.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python treq.json_content函数代码示例发布时间:2022-05-27
下一篇:
Python treq.content函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

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

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

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