本文整理汇总了Python中requests.head函数的典型用法代码示例。如果您正苦于以下问题:Python head函数的具体用法?Python head怎么用?Python head使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了head函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _check_host
def _check_host(self, host):
"""Check is host available"""
try:
requests.head(host, timeout=self.timeout)
return True
except requests.ConnectionError:
return False
开发者ID:igorer88,项目名称:series_list,代码行数:7,代码来源:base.py
示例2: CheckBundleDigest
def CheckBundleDigest(self, container, uripath, bundle_data):
"""
Download the Bundle from CloudFiles into a local path
container - the CloudFiles container in which to find the Vault DB
bundle_data - a dict containing atlest the 'id' and 'md5' of the bundle
localpath - the local path at which to store the downloaded VaultDB
"""
self.apihost = self._get_container(container)
try:
fulluri = uripath + '/BUNDLES/' + bundle_data['name']
self.ReInit(self.sslenabled, '/' + fulluri)
self.headers['X-Auth-Token'] = self.authenticator.AuthToken
self.log.debug('uri: %s', self.Uri)
self.log.debug('headers: %s', self.Headers)
try:
res = requests.head(self.Uri, headers=self.Headers)
except requests.exceptions.SSLError as ex:
self.log.error('Requests SSLError: {0}'.format(str(ex)))
res = requests.head(self.Uri, headers=self.Headers, verify=False)
if res.status_code == 404:
raise UserWarning('Server failed to find the specified bundle')
elif res.status_code >= 300:
raise UserWarning('Server responded unexpectedly during download (Code: ' + str(res.status_code) + ' )')
else:
digest = res.headers['etag'].upper()
result = (digest == bundle_data['md5'])
self.log.debug('CloudFiles Bundle Digest (' + digest + ') == Bundle MD5 (' + bundle_data['md5'] + ')? ' + str(result))
return result
except LookupError:
raise UserWarning('Invalid VaultDB Data provided.')
开发者ID:rackerlabs,项目名称:python-cloudbackup-sdk,代码行数:30,代码来源:files.py
示例3: fetch_metadata
def fetch_metadata(self):
"""
Do a HEAD request on self.url to try to get metadata
(self.length and self.mimetype).
Note that while this method fills in those attributes, it does *not*
call self.save() - so be sure to do so after calling this method!
"""
if not self.url:
return
try:
response = requests.head(self.url, timeout=5)
if response.status_code == 302:
response = requests.head(response.headers['location'],
timeout=5)
except Exception:
pass
else:
if response.status_code != 200:
return
self.length = response.headers.get('content-length')
self.mimetype = response.headers.get('content-type', '')
if self.mimetype in ('application/octet-stream', ''):
# We got a not-useful MIME type; guess!
guess = mimetypes.guess_type(self.url)
if guess[0] is not None:
self.mimetype = guess[0]
开发者ID:pculture,项目名称:django-vidscraper,代码行数:29,代码来源:models.py
示例4: __init__
def __init__(self, cfg=None, service_url=None, service_mode=None):
"""
:param cfg: Configuration for the service
:param service_url: The service URL
:param service_mode: set to "private" if imeji instance runs in "private" mode
(any other value considered as standard imeji instance mode )
If the imeji instance is not available or does not run, the instantiation will throw an error message.
"""
self.cfg = cfg or Config()
self.service_url = service_url or self.cfg.get('service', 'url')
self.service_mode_private = False or (self.cfg.get('service', 'mode', 'public') == 'private' or service_mode == 'private')
self.service_unavailable_message = \
"WARNING : The REST Interface of Imeji at {rest_service} is not available or there is another problem, " \
"check if the service is running under {imeji_service}" \
.format(imeji_service=self.service_url, rest_service=self.service_url + '/rest')
# check if Imeji instance is running and notify the user
try:
requests.head(self.service_url)
except Exception as e:
raise ImejiError(self.service_unavailable_message, e)
user = self.cfg.get('service', 'user', default=None)
password = self.cfg.get('service', 'password', default=None)
self.session = requests.Session()
if user and password:
self.session.auth = (user, password)
# initialize the request query
self.total_number_of_results = self.number_of_results = self.offset = self.size = None
开发者ID:MPDL,项目名称:pyimeji,代码行数:31,代码来源:api.py
示例5: __init__
def __init__(self, server, userAgent=_getUserAgent()):
self.server = server
self.userAgent = userAgent
try:
requests.head(self.server)
except requests.exceptions.ConnectionError, e:
raise ValueError("server %s does not look to be alive: %s" % (server, e.message))
开发者ID:wikier,项目名称:ldpy,代码行数:7,代码来源:client.py
示例6: find_url_title
def find_url_title(self, url):
"""Retrieve the title of a given URL"""
headers = {'User-Agent': 'Wget/1.13.4 (linux-gnu)'}
if url.find("://") == -1:
url = "http://" + url
try:
# a HEAD first to thwart attacks
requests.head(url, headers=headers, timeout=5)
# now the actual request
resp = requests.get(url, headers=headers)
html = resp.text
except requests.RequestException as e:
self.logger.warning(e)
return url, e.__doc__
except ValueError as e:
self.logger.warning(e)
return url, "Failed to parse url"
else:
resp.close()
cmphtml = html.lower()
start = cmphtml.find("<title")
end = cmphtml.find("</title>")
if start == -1 or end == -1:
return resp.url, "Could not find page title!"
else:
str.find
html = html[start+7:end]
html = html[html.find('>')+1:]
return resp.url, html.strip()
开发者ID:freestyl3r,项目名称:fidibot,代码行数:29,代码来源:urlparser.py
示例7: request
def request(self, url, request_method=None, auth_method=None, timeout=None, post_data=None,
user=None, password=None):
timeout = float(timeout)
try:
if request_method == "1":
# GET
if (user or password) and auth_method == "2":
req = requests.get(url, auth=(user, password), timeout=timeout, verify=False)
else:
req = requests.get(url, timeout=timeout, verify=False)
elif request_method == "2":
# POST
if (user or password) and auth_method == "2":
req = requests.post(url, data=post_data, auth=(user, password), timeout=timeout, verify=False)
else:
req = requests.post(url, data=post_data, timeout=timeout, verify=False)
elif request_method == "3":
# HEAD
if (user or password) and auth_method == "2":
req = requests.head(url, auth=(user, password), timeout=timeout, verify=False)
else:
req = requests.head(url, timeout=timeout, verify=False)
time = req.elapsed
except Exception as e:
logging.error(e)
raise
try:
code = req.status_code
response_time = time.microseconds / 1000
except Exception as e:
logging.error(e)
raise
data = [int(code), float(response_time)]
return data
开发者ID:PaesslerAG,项目名称:PythonMiniProbe,代码行数:34,代码来源:http.py
示例8: EXTRACT_MP3_LINKS
def EXTRACT_MP3_LINKS(url):
header = {'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:32.0) Gecko/20100101 Firefox/32.0',}
page = requests.get(url,header)
soup = BeautifulSoup(page.content)
links = soup.findAll("a")
#print type (links)
for item in links:
try:
if ".mp3" in item['href']:
try:
response=requests.head(item['href'])
if response.headers['content-type']=='audio/mpeg':
SAVEMP3(item['href'])
except:
pass
try:
response=requests.head(url+item['href'])
#print response
if response.headers['content-type']=='audio/mpeg':
SAVEMP3(url+item['href'])
except:
pass
except:
pass
开发者ID:shashi1920,项目名称:SongDownloader,代码行数:25,代码来源:song+downloader.py
示例9: correct_url
def correct_url(url, rights):
"""
correct_url
link checker and guesser for wikipedia thunbnail URLs
returns a checked (good) URL as a unicode string or None
"""
urlres = requests.head(url, allow_redirects=True)
# thubmnail URL looks good (check the link first)
if (urlres.status_code == requests.codes.ok):
return url
# something is not right
# if the attribute page for the image does not exist, then we
# won't find a thumbnail, so we may as well give up now
rightsres = requests.head(rights)
if (rightsres.status_code != requests.codes.ok):
return None
# okay, there should be a good thumbnail here, just not at the
# URL we tried
elif (urlres.status_code == 404):
return correct_url_404(url)
elif (urlres.status_code == 500):
return correct_url_500(url)
# not sure we can get here, something might be very wrong
else:
raise Exception("wikipedia thumbnail URL {0} had unexpected" +
"status code {1}".format(urlres.status_code, url))
开发者ID:snac-cooperative,项目名称:snac-external-search,代码行数:31,代码来源:related_links.py
示例10: download_vid
def download_vid(vid):
html_downloader = main.Downloader()
html_downloader.get(vid[1], vid[2], "html")
flv_redirect = app.extract_flv_redirect(vid[2])
headers = {
"Accept-Encoding": "gzip,deflate,sdch",
"Host": "redirector.googlevideo.com",
"Accept-Language": "en-US,en;q=0.8,fr;q=0.6",
"User-Agent": "Mozilla/5.0 (X11; Linux i686) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/30.0.1599.66 ari/537.36",
"Accept": "*/*",
"Referer": vid[1],
"Connection": "keep-alive",
"Cache-Control": "no-cache",
}
req = requests.head(flv_redirect, headers=headers)
file_url = req.headers["location"]
req = requests.head(flv_redirect, headers=headers)
file_url = req.headers["location"]
host = urlparse(file_url).netloc
headers["Host"] = host
html_downloader.download_file(file_url, "flv/%s.flv" % vid[2], headers)
开发者ID:pleasedontbelong,项目名称:xdvideos,代码行数:27,代码来源:run.py
示例11: VerifyPath
def VerifyPath(data):
# Insert try and catch blocks
try:
token_name = data["project"]["token_name"]
except:
token_name = data["project"]["project_name"]
channel_names = data["channels"].keys()
for i in range(0, len(channel_names)):
channel_type = data["channels"][channel_names[i]]["channel_type"]
path = data["channels"][channel_names[i]]["data_url"]
if channel_type == "timeseries":
timerange = data["dataset"]["timerange"]
for j in xrange(timerange[0], timerange[1] + 1):
# Test for tifs or such? Currently test for just not empty
work_path = "{}{}/{}/time{}/".format(path, token_name, channel_names[i], j)
resp = requests.head(work_path)
assert resp.status_code == 200
else:
# Test for tifs or such? Currently test for just not empty
work_path = "{}{}/{}/".format(path, token_name, channel_names[i])
resp = requests.head(work_path)
print (work_path)
assert resp.status_code == 200
开发者ID:felipebetancur,项目名称:open-connectome,代码行数:26,代码来源:verifyjson.py
示例12: __init__
def __init__(self, spider=None, *a, **kw):
super(GetpagesfromsitemapSpider, self).__init__(*a, **kw)
try:
cnx = mysql.connector.connect(**config)
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("Something is wrong with your user name or password")
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("Database does not exists")
else:
print(err)
else:
self.spider = spider
cursor = cnx.cursor()
l = []
url = "https://channelstore.roku.com"
resp = requests.head(url + "/sitemap.xml")
if (resp.status_code != 404):
l.append(resp.url)
else:
resp = requests.head(url + "/robots.txt")
if (resp.status_code == 200):
l.append(resp.url)
self.sitemap_urls = l
print self.sitemap_urls
开发者ID:dataisbeautiful,项目名称:scrapy-roku,代码行数:25,代码来源:getpagesfromsitemap.py
示例13: _check_host
def _check_host(self, host):
"""Check is host available"""
try:
requests.head(host, **self.request_params)
return True
except requests.ConnectionError:
return False
开发者ID:diannt,项目名称:series_list,代码行数:7,代码来源:base.py
示例14: has_internet
def has_internet():
"""Uses www.google.com to check connectivity"""
try:
requests.head('http://www.google.com', timeout=1)
return True
except requests.ConnectionError:
return False
开发者ID:rlee287,项目名称:pyautoupdate,代码行数:7,代码来源:pytest_skipif.py
示例15: forward
def forward(resource, identifier):
""" Redirects request for file to direct URL.
Requires global "paths" dictionary is active.
resource: a given resource, like "recount2"
identifier: relative path to file or directory
Return value: Flask redirect response object
"""
# Log all requests, even weird ones
ip = str(request.headers.get('X-Forwarded-For',
request.remote_addr)).split(',')[0].strip()
print >>_LOGSTREAM, '\t'.join(
[time.strftime('%A, %b %d, %Y at %I:%M:%S %p %Z'),
str(mmh3.hash128(ip + 'recountsalt')),
resource,
identifier])
_LOGSTREAM.flush()
if resource == 'recount':
# Redirect to IDIES URL in order of descending version
for i in ['2']: # add versions to precede 2 as they are released
if identifier.startswith(' '.join(['v', i, '/'])):
idies_url = '/'.join(
['http://idies.jhu.edu/recount/data', identifier]
)
idies_response = requests.head(idies_url)
if idies_response.status_code == 200:
return redirect(idies_url, code=302)
# v1 is not explicitly versioned
idies_url = '/'.join(['http://idies.jhu.edu/recount/data', identifier])
idies_response = requests.head(idies_url)
if idies_response.status_code == 200:
return redirect(idies_url, code=302)
abort(404)
开发者ID:nellore,项目名称:duffel,代码行数:35,代码来源:duffel.py
示例16: test_resources_available
def test_resources_available(self):
blueprint_name = 'openstack-blueprint.yaml'
self.repo_dir = clone(self.repo_url, self.workdir)
self.blueprint_yaml = self.repo_dir / blueprint_name
self.upload_blueprint(self.test_id)
invalid_resource_url = 'http://{0}/resources/blueprints/{1}/{2}' \
.format(self.env.management_ip, self.test_id, blueprint_name)
try:
result = requests.head(invalid_resource_url)
self.assertNotEqual(
result.status_code, 200,
"Resources are available through a different port than 53229.")
except ConnectionError:
pass
valid_resource_url = 'http://{0}:53229/blueprints/{1}/{2}' \
.format(self.env.management_ip, self.test_id, blueprint_name)
try:
result = requests.head(valid_resource_url)
self.assertEqual(
result.status_code, 200,
"Resources are not available through the port 53229.")
except ConnectionError:
self.fail("Resources are not available through the port 53229.")
self.cfy.delete_blueprint(self.test_id)
开发者ID:GigaSpaces-ProfessionalServices,项目名称:cloudify-system-tests,代码行数:30,代码来源:resources_available_test.py
示例17: get_video_redirect_info
def get_video_redirect_info(tag, format_, hd=False):
assert tag
assert format_ in ('webm', 'mp4'), format_
if hd:
format_ = 'hd_{}'.format(format_)
vidly_url = 'https://vid.ly/{}?content=video&format={}'.format(
tag,
format_
)
req = requests.head(vidly_url)
if req.status_code == 404:
raise VidlyNotFoundError(tag)
assert req.status_code == 302, (req.status_code, vidly_url)
req2 = requests.head(req.headers['Location'])
try:
content_length = int(req2.headers['Content-Length'])
except KeyError:
raise VideoError(
'Redirect URL lacks a Content-Length '
'(tag:{} url:{} location:{} status:{})'.format(
tag,
vidly_url,
req.headers['Location'],
req.status_code,
)
)
data = {
'url': req.headers['Location'].split('?t=')[0],
'length': content_length,
'type': req2.headers['Content-Type'],
}
return data
开发者ID:akatsoulas,项目名称:airmozilla,代码行数:32,代码来源:vidly.py
示例18: test_doesnt_delete_shared_files
def test_doesnt_delete_shared_files(self):
"""
Make sure that a file shared between two file objects doesn't
get deleted when one of the file objects gets deleted
"""
c = _create_expired_contentnode()
file_on_disk = ContentFile("test")
f = File.objects.create(
contentnode_id=c.pk,
file_on_disk=file_on_disk,
checksum="aaa",
)
f.file_on_disk.save("aaa.jpg", file_on_disk)
file_url = f.file_on_disk.url
c2 = ContentNode.objects.create(kind_id=content_kinds.TOPIC, title="test")
f2 = File.objects.create(
contentnode_id=c2.pk,
file_on_disk=file_on_disk,
checksum="aaa",
)
f2.file_on_disk.save("aaa.jpg", file_on_disk)
# check that file_url exists before cleaning up
requests.head(file_url).raise_for_status()
clean_up_contentnodes()
# the file should still be available
response = requests.head(file_url)
assert response.status_code == 200
开发者ID:fle-internal,项目名称:content-curation,代码行数:30,代码来源:test_garbage_collect.py
示例19: try_earlier_webarchivals
def try_earlier_webarchivals(webarchive_url):
'''A webarchive_url turns out to be a bad redirect. So try and earlier
webarchive of it to find what was there before the bad redirect.
Returns tuple: (result_string, good_url or None)
'''
attempts = 0
while True:
# request the link but without the redirect, revealing the earlier
# webarchivals in the headers
print webarchive_url + ' (no redirects)'
try:
req = requests.head(webarchive_url, headers=USER_AGENT, verify=False, allow_redirects=False)
except Exception, e:
return ('!! Problem with request %s' % e, None)
attempts += 1
# Link header has the options
# e.g. from http://webarchive.nationalarchives.gov.uk/+/http://www.dft.gov.uk/statistics/releases/accessibility-2010
# Link: <http://webarchive.nationalarchives.gov.uk/20140508043011/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="memento"; datetime="Thu, 08 May 2014 04:30:11 GMT", <http://webarchive.nationalarchives.gov.uk/20110826141806/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="first memento"; datetime="Fri, 26 Aug 2011 14:18:06 GMT", <http://webarchive.nationalarchives.gov.uk/20140508043011/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="last memento"; datetime="Thu, 08 May 2014 04:30:11 GMT", <http://webarchive.nationalarchives.gov.uk/20140109163921/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="prev memento"; datetime="Thu, 09 Jan 2014 16:39:21 GMT", <http://webarchive.nationalarchives.gov.uk/20140508043011/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="next memento"; datetime="Thu, 08 May 2014 04:30:11 GMT", <http://webarchive.nationalarchives.gov.uk/timegate/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="timegate", <http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="original", <http://webarchive.nationalarchives.gov.uk/timemap/http://www.dft.gov.uk/statistics/releases/accessibility-2010>; rel="timemap"; type="application/link-format"
links = req.headers['Link']
prev_links = [l.split('; ') for l in links.split(', <')
if 'rel="prev memento"' in l]
if not prev_links:
if attempts == 1:
return ('No previous webarchive links', None)
return ('No luck after trying %i previous webarchive links' % attempts, None)
webarchive_url = prev_links[0][0].strip('<>')
# Request the previous url to see if it is archived ok, or whether it
# still redirects out
print webarchive_url
try:
req = requests.head(webarchive_url, headers=USER_AGENT, verify=False)
except Exception, e:
return ('!! Problem with request %s' % e, None)
开发者ID:ArunEG,项目名称:ckanext-dgu,代码行数:34,代码来源:national_archive.py
示例20: get_tested_azure_url
def get_tested_azure_url(image_info):
try:
# tk = TimeKeeper()
# tk.time_now(image_info['flickr_id'] + '_azure_start', print_out=True)
azure_url_part = u"http://blmc.blob.core.windows.net/{0[date]}/{0[book_identifier]}_{0[volume]}_{0[page]}_{0[image_idx]}_{0[date]}_imagesize.jpg".format(image_info)
azure_url_part = azure_url_part.replace('imagesize', 'embellishments')
r = requests.head(azure_url_part, stream=True, timeout=0.3)
# tk.time_now(image_info['flickr_id'] + '_azure_embellishments', print_out=True)
if r.status_code is requests.codes.ok:
return azure_url_part
else:
azure_url_part = azure_url_part.replace('embellishments', 'medium')
r = requests.head(azure_url_part, stream=True, timeout=0.3)
# tk.time_now(image_info['flickr_id'] + '_azure_medium', print_out=True)
if r.status_code is requests.codes.ok:
return azure_url_part
else:
azure_url_part = azure_url_part.replace('medium', 'plates')
r = requests.head(azure_url_part, stream=True, timeout=0.3)
# tk.time_now(image_info['flickr_id'] + '_azure_medium', print_out=True)
if r.status_code is requests.codes.ok:
return azure_url_part
else:
return None
except:
return None
开发者ID:CSCSI,项目名称:Lost-Visions,代码行数:29,代码来源:db_tools.py
注:本文中的requests.head函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论