def __init__(self, url, absolute=False):
status = '303 See Other'
newloc = url
home = web.ctx.environ['HTTP_ORIGIN']
newloc = urlparse.urljoin(home, url)
logger.info('seeother: %s', newloc)
headers = {
'Content-Type': 'text/html',
'Location': newloc
}
web.webapi.HTTPError.__init__(self, status, headers, "")
pass
python类HTTPError()的实例源码
def GET(self, *args, **kwargs):
"""
"""
response = Response.internal_error()
try:
self.log_request()
with models.Session() as orm:
web.ctx.orm = orm
response = self.get(*args, **kwargs)
return response
except:
logger.exception('BaseHandler failure:')
status = '500 InternalError'
headers = {'Content-Type': 'text/html'}
raise web.HTTPError(status, headers, 'internal error')
def load_sqla(handler):
web.ctx.orm = scoped_session(sessionmaker(bind=engine))
try:
return handler()
except web.HTTPError:
web.ctx.orm.commit()
raise
except:
web.ctx.orm.rollback()
raise
finally:
web.ctx.orm.commit()
def get(self, bucket_name):
prefix = self.get_argument("prefix", u"")
marker = self.get_argument("marker", u"")
max_keys = int(self.get_argument("max-keys", 50000))
path = os.path.abspath(os.path.join(self.application.directory,
bucket_name))
terse = int(self.get_argument("terse", 0))
if not path.startswith(self.application.directory) or \
not os.path.isdir(path):
raise web.HTTPError(404)
object_names = []
for root, dirs, files in os.walk(path):
for file_name in files:
object_names.append(os.path.join(root, file_name))
skip = len(path) + 1
for i in range(self.application.bucket_depth):
skip += 2 * (i + 1) + 1
object_names = [n[skip:] for n in object_names]
object_names.sort()
contents = []
start_pos = 0
if marker:
start_pos = bisect.bisect_right(object_names, marker, start_pos)
if prefix:
start_pos = bisect.bisect_left(object_names, prefix, start_pos)
truncated = False
for object_name in object_names[start_pos:]:
if not object_name.startswith(prefix):
break
if len(contents) >= max_keys:
truncated = True
break
object_path = self._object_path(bucket_name, object_name)
c = {"Key": object_name}
if not terse:
info = os.stat(object_path)
c.update({
"LastModified": datetime.datetime.utcfromtimestamp(
info.st_mtime),
"Size": info.st_size,
})
contents.append(c)
marker = object_name
self.render_xml({"ListBucketResult": {
"Name": bucket_name,
"Prefix": prefix,
"Marker": marker,
"MaxKeys": max_keys,
"IsTruncated": truncated,
"Contents": contents,
}})