本文整理汇总了Python中swift.common.http.is_redirection函数的典型用法代码示例。如果您正苦于以下问题：Python is_redirection函数的具体用法？Python is_redirection怎么用？Python is_redirection使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了is_redirection函数的11个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: handle_object
def handle_object(self, env, start_response):
"""
Handles a possible static web request for an object. This object could
resolve into an index or listing request.
:param env: The original WSGI environment dict.
:param start_response: The original WSGI start_response hook.
"""
tmp_env = dict(env)
tmp_env['HTTP_USER_AGENT'] = \
'%s StaticWeb' % env.get('HTTP_USER_AGENT')
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if is_success(status_int) or is_redirection(status_int):
start_response(self._response_status, self._response_headers,
self._response_exc_info)
return resp
if status_int != HTTP_NOT_FOUND:
return self._error_response(resp, env, start_response)
self._get_container_info(env)
if not self._listings and not self._index:
return self.app(env, start_response)
status_int = HTTP_NOT_FOUND
if self._index:
tmp_env = dict(env)
tmp_env['HTTP_USER_AGENT'] = \
'%s StaticWeb' % env.get('HTTP_USER_AGENT')
if tmp_env['PATH_INFO'][-1] != '/':
tmp_env['PATH_INFO'] += '/'
tmp_env['PATH_INFO'] += self._index
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if is_success(status_int) or is_redirection(status_int):
if env['PATH_INFO'][-1] != '/':
resp = HTTPMovedPermanently(
location=env['PATH_INFO'] + '/')
self._log_response(env, resp.status_int)
return resp(env, start_response)
start_response(self._response_status, self._response_headers,
self._response_exc_info)
return resp
if status_int == HTTP_NOT_FOUND:
if env['PATH_INFO'][-1] != '/':
tmp_env = make_pre_authed_env(env, 'GET',
'/%s/%s/%s' % (self.version, self.account,
self.container),
self.agent)
tmp_env['QUERY_STRING'] = 'limit=1&format=json&delimiter' \
'=/&limit=1&prefix=%s' % quote(self.obj + '/')
resp = self._app_call(tmp_env)
body = ''.join(resp)
if not is_success(self._get_status_int()) or not body or \
not json.loads(body):
resp = HTTPNotFound()(env, self._start_response)
return self._error_response(resp, env, start_response)
resp = HTTPMovedPermanently(location=env['PATH_INFO'] +
'/')
self._log_response(env, resp.status_int)
return resp(env, start_response)
return self._listing(env, start_response, self.obj)
开发者ID:VictorLowther，项目名称:swift，代码行数:60，代码来源:staticweb.py
示例2: handle_container
def handle_container(self, env, start_response):
"""
Handles a possible static web request for a container.
:param env: The original WSGI environment dict.
:param start_response: The original WSGI start_response hook.
"""
self._get_container_info(env)
if not self._listings and not self._index:
if config_true_value(env.get('HTTP_X_WEB_MODE', 'f')):
return HTTPNotFound()(env, start_response)
return self.app(env, start_response)
if env['PATH_INFO'][-1] != '/':
resp = HTTPMovedPermanently(
location=(env['PATH_INFO'] + '/'))
return resp(env, start_response)
if not self._index:
return self._listing(env, start_response)
tmp_env = dict(env)
tmp_env['HTTP_USER_AGENT'] = \
'%s StaticWeb' % env.get('HTTP_USER_AGENT')
tmp_env['swift.source'] = 'SW'
tmp_env['PATH_INFO'] += self._index
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if status_int == HTTP_NOT_FOUND:
return self._listing(env, start_response)
elif not is_success(self._get_status_int()) or \
not is_redirection(self._get_status_int()):
return self._error_response(resp, env, start_response)
start_response(self._response_status, self._response_headers,
self._response_exc_info)
return resp
开发者ID:CiscoAS，项目名称:swift，代码行数:33，代码来源:staticweb.py
示例3: handle_container
def handle_container(self, env, start_response):
"""
Handles a possible static web request for a container.
:param env: The original WSGI environment dict.
:param start_response: The original WSGI start_response hook.
"""
self._get_container_info(env)
if not self._listings and not self._index:
if env.get("HTTP_X_WEB_MODE", "f").lower() in TRUE_VALUES:
return HTTPNotFound()(env, start_response)
return self.app(env, start_response)
if env["PATH_INFO"][-1] != "/":
resp = HTTPMovedPermanently(location=(env["PATH_INFO"] + "/"))
self._log_response(env, resp.status_int)
return resp(env, start_response)
if not self._index:
return self._listing(env, start_response)
tmp_env = dict(env)
tmp_env["HTTP_USER_AGENT"] = "%s StaticWeb" % env.get("HTTP_USER_AGENT")
tmp_env["PATH_INFO"] += self._index
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if status_int == HTTP_NOT_FOUND:
return self._listing(env, start_response)
elif not is_success(self._get_status_int()) or not is_redirection(self._get_status_int()):
return self._error_response(resp, env, start_response)
start_response(self._response_status, self._response_headers, self._response_exc_info)
return resp
开发者ID:ngtuna，项目名称:swift，代码行数:29，代码来源:staticweb.py
示例4: is_good_source
def is_good_source(self, src):
"""
Indicates whether or not the request made to the backend found
what it was looking for.
:param src: the response from the backend
:returns: True if found, False if not
"""
return is_success(src.status) or is_redirection(src.status)
开发者ID:sa4250mnpo70，项目名称:swift，代码行数:9，代码来源:base.py
示例5: _timing_stats
def _timing_stats(ctrl, *args, **kwargs):
start_time = time.time()
resp = func(ctrl, *args, **kwargs)
if is_success(resp.status_int) or is_redirection(resp.status_int) or \
resp.status_int == HTTP_NOT_FOUND:
ctrl.logger.timing_since(method + '.timing', start_time)
else:
ctrl.logger.timing_since(method + '.errors.timing', start_time)
return resp
开发者ID:sharego，项目名称:swift，代码行数:9，代码来源:utils.py
示例6: handle_container
def handle_container(self, env, start_response):
"""
Handles a possible static web request for a container.
:param env: The original WSGI environment dict.
:param start_response: The original WSGI start_response hook.
"""
container_info = self._get_container_info(env)
req = Request(env)
req.acl = container_info['read_acl']
# we checked earlier that swift.authorize is set in env
aresp = env['swift.authorize'](req)
if aresp:
resp = aresp(env, self._start_response)
return self._error_response(resp, env, start_response)
if not self._listings and not self._index:
if config_true_value(env.get('HTTP_X_WEB_MODE', 'f')):
return HTTPNotFound()(env, start_response)
return self.app(env, start_response)
if not env['PATH_INFO'].endswith('/'):
resp = HTTPMovedPermanently(
location=(env['PATH_INFO'] + '/'))
return resp(env, start_response)
if not self._index:
return self._listing(env, start_response)
tmp_env = dict(env)
tmp_env['HTTP_USER_AGENT'] = \
'%s StaticWeb' % env.get('HTTP_USER_AGENT')
tmp_env['swift.source'] = 'SW'
tmp_env['PATH_INFO'] += self._index
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if status_int == HTTP_NOT_FOUND:
return self._listing(env, start_response)
elif not is_success(self._get_status_int()) and \
not is_redirection(self._get_status_int()):
return self._error_response(resp, env, start_response)
start_response(self._response_status, self._response_headers,
self._response_exc_info)
return resp
开发者ID:HoratiusTang，项目名称:swift，代码行数:41，代码来源:staticweb.py
示例7: GETorHEAD_base
def GETorHEAD_base(self, req, server_type, partition, nodes, path,
attempts):
"""
Base handler for HTTP GET or HEAD requests.
:param req: webob.Request object
:param server_type: server type
:param partition: partition
:param nodes: nodes
:param path: path for the request
:param attempts: number of attempts to try
:returns: webob.Response object
"""
statuses = []
reasons = []
bodies = []
source = None
sources = []
newest = req.headers.get('x-newest', 'f').lower() in TRUE_VALUES
nodes = iter(nodes)
while len(statuses) < attempts:
try:
node = nodes.next()
except StopIteration:
break
if self.error_limited(node):
continue
try:
with ConnectionTimeout(self.app.conn_timeout):
headers = dict(req.headers)
headers['Connection'] = 'close'
conn = http_connect(node['ip'], node['port'],
node['device'], partition, req.method, path,
headers=headers,
query_string=req.query_string)
with Timeout(self.app.node_timeout):
possible_source = conn.getresponse()
# See NOTE: swift_conn at top of file about this.
possible_source.swift_conn = conn
except (Exception, Timeout):
self.exception_occurred(node, server_type,
_('Trying to %(method)s %(path)s') %
{'method': req.method, 'path': req.path})
continue
if possible_source.status == HTTP_INSUFFICIENT_STORAGE:
self.error_limit(node)
continue
if is_success(possible_source.status) or \
is_redirection(possible_source.status):
# 404 if we know we don't have a synced copy
if not float(possible_source.getheader('X-PUT-Timestamp', 1)):
statuses.append(HTTP_NOT_FOUND)
reasons.append('')
bodies.append('')
possible_source.read()
continue
if newest:
if sources:
ts = float(source.getheader('x-put-timestamp') or
source.getheader('x-timestamp') or 0)
pts = float(
possible_source.getheader('x-put-timestamp') or
possible_source.getheader('x-timestamp') or 0)
if pts > ts:
sources.insert(0, possible_source)
else:
sources.append(possible_source)
else:
sources.insert(0, possible_source)
source = sources[0]
statuses.append(source.status)
reasons.append(source.reason)
bodies.append('')
continue
else:
source = possible_source
break
statuses.append(possible_source.status)
reasons.append(possible_source.reason)
bodies.append(possible_source.read())
if is_server_error(possible_source.status):
self.error_occurred(node, _('ERROR %(status)d %(body)s ' \
'From %(type)s Server') %
{'status': possible_source.status,
'body': bodies[-1][:1024], 'type': server_type})
if source:
if req.method == 'GET' and \
source.status in (HTTP_OK, HTTP_PARTIAL_CONTENT):
if newest:
# we need to close all hanging swift_conns
sources.pop(0)
for src in sources:
self.close_swift_conn(src)
res = Response(request=req, conditional_response=True)
res.app_iter = self._make_app_iter(node, source, res)
# See NOTE: swift_conn at top of file about this.
res.swift_conn = source.swift_conn
update_headers(res, source.getheaders())
# Used by container sync feature
#.........这里部分代码省略.........
开发者ID:missall，项目名称:swift，代码行数:101，代码来源:base.py
示例8: is_good_source
def is_good_source(self, src):
"""
Indicates whether or not the request made to the backend found
what it was looking for.
"""
return is_success(src.status) or is_redirection(src.status)
开发者ID:CiscoAS，项目名称:swift，代码行数:6，代码来源:base.py
示例9: handle_object
def handle_object(self, env, start_response):
"""
Handles a possible static web request for an object. This object could
resolve into an index or listing request.
:param env: The original WSGI environment dict.
:param start_response: The original WSGI start_response hook.
"""
tmp_env = dict(env)
tmp_env['HTTP_USER_AGENT'] = \
'%s StaticWeb' % env.get('HTTP_USER_AGENT')
tmp_env['swift.source'] = 'SW'
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
self._get_container_info(env)
if is_success(status_int) or is_redirection(status_int):
# Treat directory marker objects as not found
if not self._dir_type:
self._dir_type = 'application/directory'
content_length = self._response_header_value('content-length')
content_length = int(content_length) if content_length else 0
if self._response_header_value('content-type') == self._dir_type \
and content_length <= 1:
status_int = HTTP_NOT_FOUND
else:
start_response(self._response_status, self._response_headers,
self._response_exc_info)
return resp
if status_int != HTTP_NOT_FOUND:
# Retaining the previous code's behavior of not using custom error
# pages for non-404 errors.
self._error = None
return self._error_response(resp, env, start_response)
if not self._listings and not self._index:
return self.app(env, start_response)
status_int = HTTP_NOT_FOUND
if self._index:
tmp_env = dict(env)
tmp_env['HTTP_USER_AGENT'] = \
'%s StaticWeb' % env.get('HTTP_USER_AGENT')
tmp_env['swift.source'] = 'SW'
if tmp_env['PATH_INFO'][-1] != '/':
tmp_env['PATH_INFO'] += '/'
tmp_env['PATH_INFO'] += self._index
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if is_success(status_int) or is_redirection(status_int):
if env['PATH_INFO'][-1] != '/':
resp = HTTPMovedPermanently(
location=env['PATH_INFO'] + '/')
return resp(env, start_response)
start_response(self._response_status, self._response_headers,
self._response_exc_info)
return resp
if status_int == HTTP_NOT_FOUND:
if env['PATH_INFO'][-1] != '/':
tmp_env = make_pre_authed_env(
env, 'GET', '/%s/%s/%s' % (
self.version, self.account, self.container),
self.agent, swift_source='SW')
tmp_env['QUERY_STRING'] = 'limit=1&format=json&delimiter' \
'=/&limit=1&prefix=%s' % quote(self.obj + '/')
resp = self._app_call(tmp_env)
body = ''.join(resp)
if not is_success(self._get_status_int()) or not body or \
not json.loads(body):
resp = HTTPNotFound()(env, self._start_response)
return self._error_response(resp, env, start_response)
resp = HTTPMovedPermanently(location=env['PATH_INFO'] + '/')
return resp(env, start_response)
return self._listing(env, start_response, self.obj)
开发者ID:SinSiXX，项目名称:swift，代码行数:71，代码来源:staticweb.py
示例10: handle_object
def handle_object(self, env, start_response):
"""
Handles a possible static web request for an object. This object could
resolve into an index or listing request.
:param env: The original WSGI environment dict.
:param start_response: The original WSGI start_response hook.
"""
tmp_env = dict(env)
tmp_env["HTTP_USER_AGENT"] = "%s StaticWeb" % env.get("HTTP_USER_AGENT")
tmp_env["swift.source"] = "SW"
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
self._get_container_info(env)
if is_success(status_int) or is_redirection(status_int):
# Treat directory marker objects as not found
if not self._dir_type:
self._dir_type = "application/directory"
content_length = self._response_header_value("content-length")
content_length = int(content_length) if content_length else 0
if self._response_header_value("content-type") == self._dir_type and content_length <= 1:
status_int = HTTP_NOT_FOUND
else:
start_response(self._response_status, self._response_headers, self._response_exc_info)
return resp
if status_int != HTTP_NOT_FOUND:
# Retaining the previous code's behavior of not using custom error
# pages for non-404 errors.
self._error = None
return self._error_response(resp, env, start_response)
if not self._listings and not self._index:
return self.app(env, start_response)
status_int = HTTP_NOT_FOUND
if self._index:
tmp_env = dict(env)
tmp_env["HTTP_USER_AGENT"] = "%s StaticWeb" % env.get("HTTP_USER_AGENT")
tmp_env["swift.source"] = "SW"
if tmp_env["PATH_INFO"][-1] != "/":
tmp_env["PATH_INFO"] += "/"
tmp_env["PATH_INFO"] += self._index
resp = self._app_call(tmp_env)
status_int = self._get_status_int()
if is_success(status_int) or is_redirection(status_int):
if env["PATH_INFO"][-1] != "/":
resp = HTTPMovedPermanently(location=env["PATH_INFO"] + "/")
return resp(env, start_response)
start_response(self._response_status, self._response_headers, self._response_exc_info)
return resp
if status_int == HTTP_NOT_FOUND:
if env["PATH_INFO"][-1] != "/":
tmp_env = make_pre_authed_env(
env,
"GET",
"/%s/%s/%s" % (self.version, self.account, self.container),
self.agent,
swift_source="SW",
)
tmp_env["QUERY_STRING"] = "limit=1&format=json&delimiter" "=/&limit=1&prefix=%s" % quote(self.obj + "/")
resp = self._app_call(tmp_env)
body = "".join(resp)
if not is_success(self._get_status_int()) or not body or not json.loads(body):
resp = HTTPNotFound()(env, self._start_response)
return self._error_response(resp, env, start_response)
resp = HTTPMovedPermanently(location=env["PATH_INFO"] + "/")
return resp(env, start_response)
return self._listing(env, start_response, self.obj)
开发者ID:ArikaChen，项目名称:swift，代码行数:66，代码来源:staticweb.py
示例11: GETorHEAD_base
def GETorHEAD_base(self, req, server_type, partition, nodes, path, attempts):
"""
Base handler for HTTP GET or HEAD requests.
:param req: swob.Request object
:param server_type: server type
:param partition: partition
:param nodes: nodes
:param path: path for the request
:param attempts: number of attempts to try
:returns: swob.Response object
"""
statuses = []
reasons = []
bodies = []
sources = []
newest = req.headers.get("x-newest", "f").lower() in TRUE_VALUES
nodes = iter(nodes)
while len(statuses) < attempts:
try:
node = nodes.next()
except StopIteration:
break
if self.error_limited(node):
continue
try:
with ConnectionTimeout(self.app.conn_timeout):
headers = dict(req.headers)
headers["Connection"] = "close"
conn = http_connect(
node["ip"],
node["port"],
node["device"],
partition,
req.method,
path,
headers=headers,
query_string=req.query_string,
)
with Timeout(self.app.node_timeout):
possible_source = conn.getresponse()
# See NOTE: swift_conn at top of file about this.
possible_source.swift_conn = conn
except (Exception, Timeout):
self.exception_occurred(
node, server_type, _("Trying to %(method)s %(path)s") % {"method": req.method, "path": req.path}
)
continue
if is_success(possible_source.status) or is_redirection(possible_source.status):
# 404 if we know we don't have a synced copy
if not float(possible_source.getheader("X-PUT-Timestamp", 1)):
statuses.append(HTTP_NOT_FOUND)
reasons.append("")
bodies.append("")
self.close_swift_conn(possible_source)
else:
statuses.append(possible_source.status)
reasons.append(possible_source.reason)
bodies.append("")
sources.append(possible_source)
if not newest: # one good source is enough
break
else:
statuses.append(possible_source.status)
reasons.append(possible_source.reason)
bodies.append(possible_source.read())
if possible_source.status == HTTP_INSUFFICIENT_STORAGE:
self.error_limit(node)
elif is_server_error(possible_source.status):
self.error_occurred(
node,
_("ERROR %(status)d %(body)s " "From %(type)s Server")
% {"status": possible_source.status, "body": bodies[-1][:1024], "type": server_type},
)
if sources:
sources.sort(key=source_key)
source = sources.pop()
for src in sources:
self.close_swift_conn(src)
res = Response(request=req, conditional_response=True)
if req.method == "GET" and source.status in (HTTP_OK, HTTP_PARTIAL_CONTENT):
res.app_iter = self._make_app_iter(node, source)
# See NOTE: swift_conn at top of file about this.
res.swift_conn = source.swift_conn
res.status = source.status
update_headers(res, source.getheaders())
if not res.environ:
res.environ = {}
res.environ["swift_x_timestamp"] = source.getheader("x-timestamp")
res.accept_ranges = "bytes"
res.content_length = source.getheader("Content-Length")
if source.getheader("Content-Type"):
res.charset = None
res.content_type = source.getheader("Content-Type")
return res
return self.best_response(req, statuses, reasons, bodies, "%s %s" % (server_type, req.method))
开发者ID:nitishb，项目名称:swift，代码行数:96，代码来源:base.py
注：本文中的swift.common.http.is_redirection函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论