本文整理汇总了Python中requests.request函数的典型用法代码示例。如果您正苦于以下问题:Python request函数的具体用法?Python request怎么用?Python request使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了request函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: perform_request
def perform_request(method, url, data_or_params=None, *args, **kwargs):
if data_or_params is None:
data_or_params = {}
headers = {'X-Edx-Api-Key': settings.API_KEY}
try:
with dog_stats_api.timer('comment_client.request.time'):
if method in ['post', 'put', 'patch']:
response = requests.request(method, url, data=data_or_params, headers=headers, timeout=5)
else:
response = requests.request(method, url, params=data_or_params, headers=headers, timeout=5)
except Exception as err:
log.exception("Trying to call {method} on {url} with params {params}".format(
method=method, url=url, params=data_or_params))
# Reraise with a single exception type
raise CommentClientError(str(err))
if 200 < response.status_code < 500:
raise CommentClientError(response.text)
# Heroku returns a 503 when an application is in maintenance mode
elif response.status_code == 503:
raise CommentClientMaintenanceError(response.text)
elif response.status_code == 500:
raise CommentClientUnknownError(response.text)
else:
if kwargs.get("raw", False):
return response.text
else:
return json.loads(response.text)
开发者ID:kaheld,项目名称:edx-platform,代码行数:28,代码来源:utils.py
示例2: _handle_request
def _handle_request(self, method, url, data=None):
"""Handle actually talking out to the trakt API, logging out debug
information, raising any relevant `TraktException` Exception types,
and extracting and returning JSON data
:param method: The HTTP method we're executing on. Will be one of
post, put, delete, get
:param url: The fully qualified url to send our request to
:param data: Optional data payload to send to the API
:return: The decoded JSON response from the Trakt API
:raises TraktException: If any non-200 return code is encountered
"""
self.logger.debug('%s: %s', method, url)
HEADERS['trakt-api-key'] = CLIENT_ID
HEADERS['Authorization'] = 'Bearer {0}'.format(OAUTH_TOKEN)
self.logger.debug('headers: %s', str(HEADERS))
self.logger.debug('method, url :: %s, %s', method, url)
if method == 'get': # GETs need to pass data as params, not body
response = requests.request(method, url, params=data,
headers=HEADERS)
else:
response = requests.request(method, url, data=json.dumps(data),
headers=HEADERS)
self.logger.debug('RESPONSE [%s] (%s): %s', method, url, str(response))
if response.status_code in self.error_map:
raise self.error_map[response.status_code]()
elif response.status_code == 204: # HTTP no content
return None
json_data = json.loads(response.content.decode('UTF-8', 'ignore'))
return json_data
开发者ID:TheJake123,项目名称:PyTrakt,代码行数:30,代码来源:core.py
示例3: test_update
def test_update(self, ref=None):
ref = ref or self.new_ref()
req_ref = ref.copy()
del req_ref['id']
resp = TestResponse({
"status_code": 200,
"text": self.serialize(ref),
})
method = 'PATCH'
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.headers[method]
kwargs['data'] = self.serialize(req_ref)
requests.request(
method,
urlparse.urljoin(
self.TEST_URL,
'v3/%s/%s' % (self.collection_key, ref['id'])),
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
returned = self.manager.update(ref['id'], **parameterize(req_ref))
self.assertTrue(isinstance(returned, self.model))
for attr in ref:
self.assertEqual(
getattr(returned, attr),
ref[attr],
'Expected different %s' % attr)
开发者ID:artofwar,项目名称:stack,代码行数:28,代码来源:utils.py
示例4: test_find
def test_find(self, ref=None):
ref = ref or self.new_ref()
ref_list = [ref]
resp = TestResponse({
"status_code": 200,
"text": self.serialize(ref_list),
})
method = 'GET'
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.headers[method]
query = '?name=%s' % ref['name'] if hasattr(ref, 'name') else ''
requests.request(
method,
urlparse.urljoin(
self.TEST_URL,
'v3/%s%s' % (self.collection_key, query)),
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
returned = self.manager.find(name=getattr(ref, 'name', None))
self.assertTrue(isinstance(returned, self.model))
for attr in ref:
self.assertEqual(
getattr(returned, attr),
ref[attr],
'Expected different %s' % attr)
开发者ID:dsqmoore,项目名称:python-keystoneclient,代码行数:27,代码来源:utils.py
示例5: test_get_version_local
def test_get_version_local(self):
resp = utils.TestResponse({
"status_code": 300,
"text": json.dumps(self.TEST_RESPONSE_DICT),
})
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.TEST_REQUEST_HEADERS
requests.request('GET',
"http://localhost:35357",
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
cs = client.Client()
versions = cs.discover()
self.assertIsInstance(versions, dict)
self.assertIn('message', versions)
self.assertIn('v3.0', versions)
self.assertEquals(
versions['v3.0']['url'],
self.TEST_RESPONSE_DICT['versions']['values'][0]['links'][0]
['href'])
self.assertEquals(
versions['v2.0']['url'],
self.TEST_RESPONSE_DICT['versions']['values'][1]['links'][0]
['href'])
开发者ID:CiscoAS,项目名称:python-keystoneclient,代码行数:25,代码来源:test_discover.py
示例6: download_files
def download_files(urls, path):
url = urls['url']
size = urls['size']
fid = urls['id']
filename = fid + '_' + unquote(url).split('/')[-1]
if os.path.exists(path + '/' + filename):
if os.path.getsize(path + '/' + filename) == size:
succ_list.append(url)
return
print "Downloading... :" + filename + "\tsize: " + str(size/1024.0) + "KB"
response = requests.request("GET", url, headers=headers)
if response.status_code != 200:
time.sleep(3)
response = requests.request("GET", url, headers=headers)
if response.status_code != 200:
time.sleep(3)
response = requests.request("GET", url, headers=headers)
if response.status_code != 200:
print "ERROR", url
err_list.append(url)
return
with open(path + "/" + filename, "wb+") as f:
f.write(response.content)
succ_list.append(url)
开发者ID:WANG-lp,项目名称:basecamp-download,代码行数:30,代码来源:download.py
示例7: test_contract
def test_contract(self):
stub_definition_file = os.path.join(
os.environ['CONSUMER_CONTRACTS_ROOT'],
'contracts/includes/consumer2.json'
)
with open(stub_definition_file, 'r') as f:
stub_definition = json.load(f)
path = stub_definition["predicates"][0]["equals"]["path"]
method = stub_definition["predicates"][0]["equals"]["method"]
record = json.loads(stub_definition["responses"][0]["is"]["body"])
provider.DataStore.save_record(record)
contractual_response = requests.request(
method,
self.stub_host_port+path
)
actual_response = requests.request(method, self.actual_host_port+path)
self.assertEqual(
actual_response.status_code,
contractual_response.status_code
)
# The consumer shouldn't mind if the provider returns some
# extra data. Following Postel's law.
self.assertDictContainsSubset(
contractual_response.json(),
actual_response.json()
)
provider.DataStore.delete_record(record)
开发者ID:tdpreece,项目名称:consumer_driven_contracts_with_mountebank,代码行数:30,代码来源:test.py
示例8: load
def load(method, url, data):
if method in ['GET', 'DELETE']:
print 'FETCH %s %s' % (url, data)
response = requests.request(method, url, params=data, allow_redirects=True)
if method in ['POST', 'PUT']:
files = {}
for key in data:
if hasattr(data[key], 'read'):
files[key] = data[key]
for key in files:
data.pop(key)
print 'FETCH %s %s' % (url, data)
response = requests.request(method, url, data=data, files=files)
result = self._parse(response.content)
try:
next_url = result['paging']['next']
except (KeyError, TypeError):
next_url = None
return result, next_url
开发者ID:10clouds,项目名称:citypulse,代码行数:25,代码来源:crawler.py
示例9: test_update_own_password
def test_update_own_password(self):
req_body = {
'user': {
'password': 'ABCD', 'original_password': 'DCBA'
}
}
resp_body = {
'access': {}
}
resp = utils.TestResponse({
"status_code": 200,
"text": json.dumps(resp_body)
})
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.TEST_POST_HEADERS
kwargs['data'] = json.dumps(req_body)
requests.request(
'PATCH',
urlparse.urljoin(self.TEST_URL, 'v2.0/OS-KSCRUD/users/123'),
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
self.client.user_id = '123'
self.client.users.update_own_password('DCBA', 'ABCD')
开发者ID:drusskikh,项目名称:python-keystoneclient,代码行数:26,代码来源:test_users.py
示例10: load
def load(method, url, data):
if method in ['GET', 'DELETE']:
try:
response = requests.request(method, url, params=data, allow_redirects=True)
except requests.RequestException as exception:
raise self.HTTPError(exception.message)
if method in ['POST', 'PUT']:
files = {}
for key in data:
if hasattr(data[key], 'read'):
files[key] = data[key]
for key in files:
data.pop(key)
try:
response = requests.request(method, url, data=data, files=files)
except requests.RequestException as exception:
raise self.HTTPError(exception.message)
result = self._parse(response.content)
try:
next_url = result['paging']['next']
except (KeyError, TypeError):
next_url = None
return result, next_url
开发者ID:bunchesofdonald,项目名称:facepy,代码行数:32,代码来源:graph_api.py
示例11: request
def request(method, url, **kwargs):
try:
_L.debug("Requesting %s with args %s", url, kwargs.get('params') or kwargs.get('data'))
return requests.request(method, url, timeout=_http_timeout, **kwargs)
except requests.exceptions.SSLError as e:
_L.warning("Retrying %s without SSL verification", url)
return requests.request(method, url, timeout=_http_timeout, verify=False, **kwargs)
开发者ID:cequencer,项目名称:machine,代码行数:7,代码来源:cache.py
示例12: test_authenticate_failure
def test_authenticate_failure(self):
_auth = 'auth'
_cred = 'passwordCredentials'
_pass = 'password'
self.TEST_REQUEST_BODY[_auth][_cred][_pass] = 'bad_key'
resp = utils.TestResponse({
"status_code": 401,
"text": json.dumps({
"unauthorized": {
"message": "Unauthorized",
"code": "401",
},
}),
})
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.TEST_REQUEST_HEADERS
kwargs['data'] = json.dumps(self.TEST_REQUEST_BODY)
requests.request('POST',
self.TEST_URL + "/tokens",
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
# Workaround for issue with assertRaises on python2.6
# where with assertRaises(exceptions.Unauthorized): doesn't work
# right
def client_create_wrapper():
client.Client(username=self.TEST_USER,
password="bad_key",
tenant_id=self.TEST_TENANT_ID,
auth_url=self.TEST_URL)
self.assertRaises(exceptions.Unauthorized, client_create_wrapper)
开发者ID:CiscoAS,项目名称:python-keystoneclient,代码行数:33,代码来源:test_auth.py
示例13: test_authenticate_success_token_scoped
def test_authenticate_success_token_scoped(self):
del self.TEST_REQUEST_BODY['auth']['passwordCredentials']
self.TEST_REQUEST_BODY['auth']['token'] = {'id': self.TEST_TOKEN}
self.TEST_REQUEST_HEADERS['X-Auth-Token'] = self.TEST_TOKEN
resp = utils.TestResponse({
"status_code": 200,
"text": json.dumps(self.TEST_RESPONSE_DICT),
})
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.TEST_REQUEST_HEADERS
kwargs['data'] = json.dumps(self.TEST_REQUEST_BODY)
requests.request('POST',
self.TEST_URL + "/tokens",
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
cs = client.Client(token=self.TEST_TOKEN,
tenant_id=self.TEST_TENANT_ID,
auth_url=self.TEST_URL)
self.assertEqual(cs.management_url,
self.TEST_RESPONSE_DICT["access"]["serviceCatalog"][3]
['endpoints'][0]["adminURL"])
self.assertEqual(cs.auth_token,
self.TEST_RESPONSE_DICT["access"]["token"]["id"])
开发者ID:CiscoAS,项目名称:python-keystoneclient,代码行数:25,代码来源:test_auth.py
示例14: _request
def _request(self, method, url_path, data):
url = urljoin(self.server, url_path)
headers = {
'Content-Type': 'application/json'
}
if self.token:
headers['Authorization'] = 'Bearer {}'.format(self.token)
if method == 'GET':
response = requests.request(
method, url, params=data, headers=headers,
verify=False)
else:
response = requests.request(
method, url, data=data, headers=headers,
verify=False)
result = json.loads(response.content.decode('utf-8'))
if response.status_code not in [200, 201]:
logger.debug(' {status}: "{message}"'.format(**result))
else:
logger.debug(' Success')
return result
开发者ID:telepenin,项目名称:test-ingress,代码行数:27,代码来源:k8s_client.py
示例15: test_authenticate_success
def test_authenticate_success(self):
TEST_TOKEN = "abcdef"
self.TEST_RESPONSE_HEADERS['X-Subject-Token'] = TEST_TOKEN
ident = self.TEST_REQUEST_BODY['auth']['identity']
del ident['password']['user']['domain']
del ident['password']['user']['name']
ident['password']['user']['id'] = self.TEST_USER
resp = utils.TestResponse({
"status_code": 200,
"text": json.dumps(self.TEST_RESPONSE_DICT),
"headers": self.TEST_RESPONSE_HEADERS,
})
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.TEST_REQUEST_HEADERS
kwargs['data'] = json.dumps(self.TEST_REQUEST_BODY, sort_keys=True)
requests.request('POST',
self.TEST_URL + "/auth/tokens",
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
cs = client.Client(user_id=self.TEST_USER,
password=self.TEST_TOKEN,
project_id=self.TEST_TENANT_ID,
auth_url=self.TEST_URL)
self.assertEqual(cs.auth_token, TEST_TOKEN)
开发者ID:brandon-adams,项目名称:python-keystoneclient,代码行数:26,代码来源:test_auth.py
示例16: prod_api
def prod_api(token):
non_gzipped_headers = {
'Authorization': token,
'Accept': 'application/json, */*'
}
gzipped_headers = {
'Authorization': token,
'Accept': 'application/json, */*',
'Accept-Encoding': 'gzip'
}
urls = [
"/api/gis/coordinates/",
"/api/gis/county_boundaries/",
"/api/gis/ward_boundaries/",
"/api/gis/constituency_boundaries/",
"/api/common/filtering_summaries/"
"/api/facilities/facilities/",
"/api/facilities/facilities_list/",
"/api/facilities/facilities_list/?format=excel"
]
for i in urls:
# warmup non-gzip encoded content
requests.request(
"GET", url=_get_url(i), headers=non_gzipped_headers
)
# warmup gzip encoded content
requests.request("GET", url=_get_url(i), headers=gzipped_headers)
开发者ID:jmusembi,项目名称:mfl_api,代码行数:28,代码来源:fabfile.py
示例17: test_authenticate_failure
def test_authenticate_failure(self):
ident = self.TEST_REQUEST_BODY['auth']['identity']
ident['password']['user']['password'] = 'bad_key'
resp = utils.TestResponse({
"status_code": 401,
"text": json.dumps({
"unauthorized": {
"message": "Unauthorized",
"code": "401",
},
}),
})
kwargs = copy.copy(self.TEST_REQUEST_BASE)
kwargs['headers'] = self.TEST_REQUEST_HEADERS
kwargs['data'] = json.dumps(self.TEST_REQUEST_BODY, sort_keys=True)
requests.request('POST',
self.TEST_URL + "/auth/tokens",
**kwargs).AndReturn((resp))
self.mox.ReplayAll()
# Workaround for issue with assertRaises on python2.6
# where with assertRaises(exceptions.Unauthorized): doesn't work
# right
def client_create_wrapper():
client.Client(user_domain_name=self.TEST_DOMAIN_NAME,
username=self.TEST_USER,
password="bad_key",
project_id=self.TEST_TENANT_ID,
auth_url=self.TEST_URL)
self.assertRaises(exceptions.Unauthorized, client_create_wrapper)
开发者ID:brandon-adams,项目名称:python-keystoneclient,代码行数:32,代码来源:test_auth.py
示例18: api_call
def api_call(path, method='GET', query=None, data=None, token=None, flag_full_path=False):
# Translates all the HTTP calls to interface with the API
data = json.dumps(data) if isinstance(data, dict) or isinstance(data,list) else None
base_url = 'https://cgc-api.sbgenomics.com/v2/'
if token == None:
if 'AUTH_TOKEN' in os.environ.keys():
token = os.environ['AUTH_TOKEN']
else:
print("""
You have failed to set your AUTH_TOKEN, we can not continue.
Please go to set_AUTH_TOKEN.ipynb and set it properly.
""")
return False
headers = {
'X-SBG-Auth-Token': token,
'Accept': 'application/json',
'Content-type': 'application/json',
}
if flag_full_path:
response = request(method, path, params=query, data=data, headers=headers)
else:
response = request(method, base_url + path, params=query, data=data, headers=headers)
response_dict = json.loads(response.content) if response.content else {}
if response.status_code / 100 != 2:
print(response_dict['message'])
print('Error Code: %i.' % (response_dict['code']))
print(response_dict['more_info'])
raise Exception('Server responded with status code %s.' % response.status_code)
return response_dict
开发者ID:damirkrstanovic,项目名称:okAPI,代码行数:33,代码来源:apimethods.py
示例19: download_urls
def download_urls(project, count=0):
url = "https://your_domain_should_be_here.basecamphq.com/projects/%s/attachments?n=%s" % (project, count)
response = requests.request("GET", url, headers=headers)
while response.status_code != 200:
time.sleep(3)
response = requests.request("GET", url, headers=headers)
if response.status_code != 200:
print "ERROR:", project
return '', []
# print(response.text)
print response.status_code
tree = ET.fromstring(response.text)
files = []
attas = tree.findall('attachment')
for att in attas:
f = {}
f['size'] = int(att.find('byte-size').text)
f['id'] = att.find('id').text
f['url'] = att.find('download-url').text
files.append(f)
return response.text, files
开发者ID:WANG-lp,项目名称:basecamp-download,代码行数:29,代码来源:download.py
示例20: getRealLinks
def getRealLinks(self):
realLinks = []
for links in self.pageLinks:
headers = {
'cache-control': "no-cache",
'postman-token': "8af4828e-133e-5c8f-e4bd-10ac70d6048a"
}
try:
response = requests.request("GET", links, headers=headers)
except:
response = requests.request("GET", links, headers=headers)
m = re.search('(orig=.+?"http.+?")', response.text)
try:
cleanUrl = m.group(0)[m.group(0).index("http"):]
cleanUrl = cleanUrl[:cleanUrl.rindex("\\")]
except ValueError as e:
print(response.text)
print(e)
realLinks.append(cleanUrl)
return realLinks
开发者ID:sampourcyrous,项目名称:scribd-download-online-book,代码行数:25,代码来源:downloadBookOnline.py
注:本文中的requests.request函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论