python类request()的实例源码

__init__.py 文件源码 项目:mturk-crowd-beta-client-python 作者: awslabs 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def _make_request(self, method, function_name, task_name, body=None):
        uri_path = self._build_uri_path(function_name, task_name)
        headers = self._build_aws_sigv4_headers(method, uri_path, body)
        url = _ENDPOINT + uri_path

        request = Request(
            method,
            url,
            headers=headers,
            json=body).prepare()
        logger.debug('Invoking {} on URL {} with headers={}, body={}'.format(
            request.method,
            request.url,
            json.dumps(dict(request.headers)),
            request.body))
        response = Session().send(request, timeout=_TIMEOUT_SECONDS)
        return response
mockrequests.py 文件源码 项目:wallstreet 作者: mcdallas 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def prepare_request(*args, **kwargs):
    r = requests.Request(*args, **kwargs)
    s = requests.Session()
    return s.prepare_request(r)
fbarc.py 文件源码 项目:fbarc 作者: justinlittman 项目源码 文件源码 阅读 39 收藏 0 点赞 0 评论 0
def generate_url(self, node_id, definition_name, escape=False):
        """
        Returns the url for retrieving the specified node from the Graph API
        given the node type definition.
        """
        url, params = self._prepare_node_request(node_id, definition_name)
        if not escape:
            return '{}?{}'.format(url, '&'.join(['{}={}'.format(k, v) for k, v in params.items()]))
        else:
            return requests.Request('GET', url, params=params).prepare().url
models.py 文件源码 项目:pyfc4 作者: ghukill 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def start_txn(self, txn_name=None):

        '''
        Request new transaction from repository, init new Transaction, 
        store in self.txns

        Args:
            txn_name (str): human name for transaction

        Return:
            (Transaction): returns intance of newly created transaction
        '''

        # if no name provided, create one
        if not txn_name:
            txn_name = uuid.uuid4().hex

        # request new transaction
        txn_response = self.api.http_request('POST','%s/fcr:tx' % self.root, data=None, headers=None)

        # if 201, transaction was created
        if txn_response.status_code == 201:

            txn_uri = txn_response.headers['Location']
            logger.debug("spawning transaction: %s" % txn_uri)

            # init new Transaction, and pass Expires header
            txn = Transaction(
                self, # pass the repository
                txn_name,
                txn_uri,
                expires = txn_response.headers['Expires'])

            # append to self
            self.txns[txn_name] = txn

            # return 
            return txn
test_requests.py 文件源码 项目:talisker 作者: canonical-ols 项目源码 文件源码 阅读 38 收藏 0 点赞 0 评论 0
def response(
        method='GET',
        host='http://example.com',
        url='/',
        code=200,
        elapsed=1.0):
    req = requests.Request(method, host + url)
    resp = requests.Response()
    resp.request = req.prepare()
    resp.status_code = code
    resp.elapsed = timedelta(seconds=elapsed)
    return resp
test_requests.py 文件源码 项目:talisker 作者: canonical-ols 项目源码 文件源码 阅读 43 收藏 0 点赞 0 评论 0
def test_configure():
    session = requests.Session()
    talisker.requests.configure(session)

    req = requests.Request('GET', 'http://localhost')
    with talisker.request_id.context('XXX'):
        prepared = session.prepare_request(req)

    assert prepared.headers['X-Request-Id'] == 'XXX'
httpclient.py 文件源码 项目:trio2o 作者: openstack 项目源码 文件源码 阅读 33 收藏 0 点赞 0 评论 0
def forward_req(context, action, b_headers, b_url, b_body):
    s = Session()
    req = Request(action, b_url,
                  data=b_body,
                  headers=b_headers)
    prepped = req.prepare()

    # do something with prepped.body
    # do something with prepped.headers
    resp = s.send(prepped,
                  timeout=60)

    return resp
base.py 文件源码 项目:threatstack-python-client 作者: MyPureCloud 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def http_request(self, method, path, data=None, params=None):
        """ Wraps HTTP calls to ThrestStack API """

        s = Session()
        url = urljoin(self.BASE_URL, path)
        headers = {"Authorization": self.api_key}
        if self.org_id:
            headers[self.org_id_header] = self.org_id

        req = Request(
            method,
            url,
            headers=headers,
            data=data,
            params=params
        )
        prepped = req.prepare()
        resp = s.send(prepped, timeout=self.timeout)
        if resp.status_code == 429:
            raise errors.APIRateLimitError("Threat Stack API rate limit exceeded")
        else:
            return self.handle_response(resp)
dump.py 文件源码 项目:mygene.info 作者: biothings 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
def _query(self, *args, **kwargs):
        req = requests.Request(*args, **kwargs)
        res = self.client.send(req.prepare())
        if res.status_code != 200:
            raise MartException(res)
        if res.text.startswith('Query ERROR:'):
            raise MartException(res.text)
        return res.text
pycarwings2.py 文件源码 项目:pycarwings2 作者: jdhorne 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _request(self, endpoint, params):
        params["initial_app_strings"] = "geORNtsZe5I4lRGjG9GZiA"
        if self.custom_sessionid:
            params["custom_sessionid"] = self.custom_sessionid
        else:
            params["custom_sessionid"] = ""

        req = Request('POST', url=BASE_URL + endpoint, data=params).prepare()

        log.debug("invoking carwings API: %s" % req.url)
        log.debug("params: %s" % json.dumps(params, sort_keys=True, indent=3, separators=(',', ': ')))

        try:
            sess = requests.Session()
            response = sess.send(req)
            log.debug('Response HTTP Status Code: {status_code}'.format(
                status_code=response.status_code))
            log.debug('Response HTTP Response Body: {content}'.format(
                content=response.content))
        except RequestException:
            log.warning('HTTP Request failed')

        j = json.loads(response.content)

        if "message" in j and j["message"] == "INVALID PARAMS":
            log.error("carwings error %s: %s" % (j["message"], j["status"]) )
            raise CarwingsError("INVALID PARAMS")
        if "ErrorMessage" in j:
            log.error("carwings error %s: %s" % (j["ErrorCode"], j["ErrorMessage"]) )
            raise CarwingsError

        return j


问题


面经


文章

微信
公众号

扫码关注公众号