本文整理汇总了Python中wsgiref.headers.Headers类的典型用法代码示例。如果您正苦于以下问题:Python Headers类的具体用法?Python Headers怎么用?Python Headers使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Headers类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: patched_start_response
def patched_start_response(status, headers, exc_info=None):
# if self._should_handle(headers)
wsgi_headers = Headers(headers)
# If we're debugging, or the response already has an expires
# header, just skip this.
log.debug('Skipping expired headers' if self.debug else 'Calculating expires headers')
if not self.debug and 'Expires' not in wsgi_headers:
mime = wsgi_headers.get('Content-Type', '*').split(';')[0]
log.debug('See mime type ' + mime)
# If the mime type is explicitly called out, use the expire
# delay specified.
if mime in self.expire_seconds:
log.debug('Matched mimetype exactly.')
expire_time = self.make_expire_time_for(mime)
# If there's a catch-all wildcard delay, use that.
elif '*' in self.expire_seconds:
log.debug('Matched mimetype with universal.')
expire_time = self.make_expire_time_for('*')
# Otherwise, don't set the header.
else:
log.debug('No mimetype match.')
expire_time = None
if expire_time is not None:
log.debug('Adding expires header value: ' + expire_time)
headers.append(('Expires', expire_time))
log.debug('-'*60)
return start_response(status, headers, exc_info)
开发者ID:CrowdSpot,项目名称:shareabouts,代码行数:33,代码来源:twinkie.py
示例2: testMappingInterface
def testMappingInterface(self):
test = [("x", "y")]
self.assertEqual(len(Headers([])), 0)
self.assertEqual(len(Headers(test[:])), 1)
self.assertEqual(Headers(test[:]).keys(), ["x"])
self.assertEqual(Headers(test[:]).values(), ["y"])
self.assertEqual(Headers(test[:]).items(), test)
self.assertIsNot(Headers(test).items(), test) # must be copy!
h = Headers([])
del h["foo"] # should not raise an error
h["Foo"] = "bar"
for m in h.has_key, h.__contains__, h.get, h.get_all, h.__getitem__:
self.assertTrue(m("foo"))
self.assertTrue(m("Foo"))
self.assertTrue(m("FOO"))
self.assertFalse(m("bar"))
self.assertEqual(h["foo"], "bar")
h["foo"] = "baz"
self.assertEqual(h["FOO"], "baz")
self.assertEqual(h.get_all("foo"), ["baz"])
self.assertEqual(h.get("foo", "whee"), "baz")
self.assertEqual(h.get("zoo", "whee"), "whee")
self.assertEqual(h.setdefault("foo", "whee"), "baz")
self.assertEqual(h.setdefault("zoo", "whee"), "whee")
self.assertEqual(h["foo"], "baz")
self.assertEqual(h["zoo"], "whee")
开发者ID:van7hu,项目名称:fanca,代码行数:30,代码来源:test_wsgiref.py
示例3: handle_application_error
def handle_application_error(self, environ, start_response):
status = "500 Internal Server Error"
headers = Headers([])
# Package the exception info as into a special header and
# send it to the client
type, exc, tb = sys.exc_info()
tbfile = StringIO()
traceback.print_exc(file=tbfile)
headers['Content-Type'] = 'text/plain; charset=utf-8'
LOG.debug("Packing traceback context into debug header: %s",
self.debug_header)
debug_header = self.pack_header(Traceback(tb))
LOG.debug("Debug header (%d bytes): %s",
len(debug_header), debug_header)
headers[self.debug_header] = debug_header
app_uri = application_uri(environ)
headers["Location"] = app_uri[:-1] + self.debug_uri
start_response(status, headers.items())
return [tbfile.getvalue().encode('utf-8')]
开发者ID:te-je,项目名称:ohoh,代码行数:25,代码来源:middleware.py
示例4: testMappingInterface
def testMappingInterface(self):
test = [('x','y')]
self.assertEqual(len(Headers([])),0)
self.assertEqual(len(Headers(test[:])),1)
self.assertEqual(Headers(test[:]).keys(), ['x'])
self.assertEqual(Headers(test[:]).values(), ['y'])
self.assertEqual(Headers(test[:]).items(), test)
self.assertFalse(Headers(test).items() is test) # must be copy!
h=Headers([])
del h['foo'] # should not raise an error
h['Foo'] = 'bar'
for m in h.__contains__, h.get, h.get_all, h.__getitem__:
self.assertTrue(m('foo'))
self.assertTrue(m('Foo'))
self.assertTrue(m('FOO'))
self.assertFalse(m('bar'))
self.assertEqual(h['foo'],'bar')
h['foo'] = 'baz'
self.assertEqual(h['FOO'],'baz')
self.assertEqual(h.get_all('foo'),['baz'])
self.assertEqual(h.get("foo","whee"), "baz")
self.assertEqual(h.get("zoo","whee"), "whee")
self.assertEqual(h.setdefault("foo","whee"), "baz")
self.assertEqual(h.setdefault("zoo","whee"), "whee")
self.assertEqual(h["foo"],"baz")
self.assertEqual(h["zoo"],"whee")
开发者ID:524777134,项目名称:cpython,代码行数:30,代码来源:test_wsgiref.py
示例5: parse_http_headers
def parse_http_headers(environ):
h = Headers([])
for k, v in environ.items():
if k.startswith('HTTP_'):
name = k[5:]
h.add_header(name, v)
return h
开发者ID:minineko,项目名称:test1,代码行数:7,代码来源:utils.py
示例6: thread_app
def thread_app(env, resp):
path = env['PATH_INFO']
# utils.log('thread_app', path)
m = thread_re.match(path)
board, datkey = m.group(1), m.group(2)
key = keylib.get_filekey(datkey)
data = cache.Cache(key)
data.load()
if check_get_cache(env):
if not data.exists() or len(data) == 0:
# when first access, load data from network
data.search()
elif _count_is_update(key):
# update thread
# limit `data.search` calling. it's slow!
threading.Thread(target=data.search, daemon=True).start()
if not data.exists():
resp('404 Not Found', [('Content-Type', 'text/plain; charset=Shift_JIS')])
return [b'404 Not Found']
thread = dat.make_dat(data, env, board)
headers = Headers([('Content-Type', 'text/plain; charset=Shift_JIS')])
last_m = eutils.formatdate(data.stamp)
headers['Last-Modified'] = last_m
resp("200 OK", headers.items())
return (c.encode('cp932', 'replace') for c in thread)
开发者ID:disunbow,项目名称:saku,代码行数:32,代码来源:datd.py
示例7: board_app
def board_app(env, resp):
path = env['PATH_INFO']
m = board_re.match(path)
board = m.group(1)
message = gateway.search_message(env.get('HTTP_ACCEPT_LANGUAGE', 'ja'))
headers = Headers([('Content-Type', 'text/html; charset=Shift_JIS')])
resp("200 OK", headers.items())
board = utils.sanitize(utils.get_board(path))
if board:
fmt = '{logo} - {board} - {desc}'
else:
fmt = '{logo} - {desc}'
text = fmt.format(logo=message['logo'], desc=message['description'], board=board)
html = '''
<!DOCTYPE html>
<html><head>
<meta http-equiv="content-type" content="text/html; charset=Shift_JIS">
<title>{text}</title>
<meta name="description" content="{text}">
</head><body>
<h1>{text}</h1>
</body></html>
'''.format(text=text)
return [html.encode('cp932', 'replace')]
开发者ID:disunbow,项目名称:saku,代码行数:29,代码来源:datd.py
示例8: patched_start_response
def patched_start_response(status, headers, exc_info=None):
# if self._should_handle(headers)
wsgi_headers = Headers(headers)
# If we're debugging, or the response already has an expires
# header, just skip this.
if not self.debug and "Expires" not in wsgi_headers:
mime = wsgi_headers.get("Content-Type", "*").split(";")[0]
# If the mime type is explicitly called out, use the expire
# delay specified.
if mime in self.expire_seconds:
expire_time = self.make_expire_time_for(mime)
# If there's a catch-all wildcard delay, use that.
elif "*" in self.expire_seconds:
expire_time = self.make_expire_time_for("*")
# Otherwise, don't set the header.
else:
expire_time = None
if expire_time is not None:
log.debug("Adding expires header value: " + expire_time)
headers.append(("Expires", expire_time))
return start_response(status, headers, exc_info)
开发者ID:joshk,项目名称:planbox,代码行数:27,代码来源:twinkie.py
示例9: getTileResponse
def getTileResponse(self, coord, extension, ignore_cached=False):
""" Get status code, headers, and a tile binary for a given request layer tile.
Arguments:
- coord: one ModestMaps.Core.Coordinate corresponding to a single tile.
- extension: filename extension to choose response type, e.g. "png" or "jpg".
- ignore_cached: always re-render the tile, whether it's in the cache or not.
This is the main entry point, after site configuration has been loaded
and individual tiles need to be rendered.
"""
start_time = time()
mimetype, format = self.getTypeByExtension(extension)
# default response values
status_code = 200
headers = Headers([('Content-Type', mimetype)])
body = None
cache = self.config.cache
if not ignore_cached:
# Start by checking for a tile in the cache.
try:
body = cache.read(self, coord, format)
except TheTileLeftANote, e:
headers = e.headers
status_code = e.status_code
body = e.content
if e.emit_content_type:
headers.setdefault('Content-Type', mimetype)
tile_from = 'cache'
开发者ID:pjmtdw,项目名称:TileStache,代码行数:35,代码来源:Core.py
示例10: get_alternatives
def get_alternatives(base_headers, files):
# Sort by size so that the smallest compressed alternative matches first
alternatives = []
files_by_size = sorted(files.items(), key=lambda i: i[1].stat.st_size)
for encoding, file_entry in files_by_size:
headers = Headers(base_headers.items())
headers['Content-Length'] = str(file_entry.stat.st_size)
if encoding:
headers['Content-Encoding'] = encoding
encoding_re = re.compile(r'\b%s\b' % encoding)
else:
encoding_re = re.compile('')
alternatives.append((encoding_re, file_entry.path, headers.items()))
return alternatives
开发者ID:CollinsMuiruri,项目名称:Instagram,代码行数:14,代码来源:responders.py
示例11: signal_land_or_sea
def signal_land_or_sea(self, body, layer, coord, format):
if body:
md5sum = self.md5sum(body)
second_md5sum = self.md5sum(Disk.read(self, self.second, coord, format))
headers = Headers([('Access-Control-Expose-Headers', 'X-Land-Or-Sea')])
headers.setdefault('X-Land-Or-Sea', '0')
if second_md5sum and md5sum == second_md5sum:
if md5sum == self.land_md5:
headers['X-Land-Or-Sea'] = '1'
elif md5sum == self.sea_md5:
headers['X-Land-Or-Sea'] = '2'
raise TheTileLeftANote(content=body, headers=headers)
开发者ID:stamen,项目名称:tilestache-goodies,代码行数:15,代码来源:__init__.py
示例12: __call__
def __call__(self, environ, start_response):
key_morsel = Cookie(environ.get("HTTP_COOKIE", "")).get(self.toggle_key)
# useful vars
query = query_str2dict(environ.get("QUERY_STRING"))
enable_by_cookie = key_morsel.value == self.enable_value if key_morsel else False
enable_by_query = query.get(self.toggle_key) == self.enable_value
# pop toggle_key from query dic to avoid case: '?_profile=on&_profile='
disable = query.pop(self.toggle_key, None) == "" # only can be disabled by query
enable = not disable and (enable_by_query or enable_by_cookie)
run_app, resp_body, saved_ss_args = self._intercept_call()
# processing cookies and queries
so = query.pop(self.SIMPLE_OUTPUT_TOGGLE_KEY, None)
if so is not None:
self.simple_output = so == "True"
cookie_to_set = None
if enable_by_query and not enable_by_cookie:
cookie_to_set = "%s=%s; Path=/; HttpOnly" % (self.toggle_key, self.enable_value)
elif disable:
cookie_to_set = "%s=; Path=/; Max-Age=1; HttpOnly" % self.toggle_key
if enable:
start = time.time()
profile = Profile()
profile.runcall(run_app, environ) # here we call the WSGI app
elapsed = time.time() - start
else:
profile = elapsed = None # for annoying IDE
run_app(environ)
status, headers = saved_ss_args[:2]
headers_dic = Headers(headers)
if cookie_to_set:
headers_dic.add_header("Set-Cookie", cookie_to_set)
# insert result into response
content_type = headers_dic.get("Content-Type", "")
if enable and status.startswith("200") and content_type.startswith("text/html"):
environ["QUERY_STRING"] = dict2query_str(query)
matched = _find_charset.match(content_type)
encoding = matched.group(1) if matched else "ascii"
rendered = self.render_result(profile, elapsed, environ).encode(encoding, "replace")
resp_body = [insert_into_body(rendered, b"".join(resp_body))]
headers_dic["Content-Length"] = str(len(resp_body[0]))
start_response(status, headers, saved_ss_args[2] if len(saved_ss_args) == 3 else None)
return resp_body
开发者ID:krrr,项目名称:nano-wsgi-profiler,代码行数:48,代码来源:__init__.py
示例13: get_static_file
def get_static_file(self, path, url, stat_cache=None):
# Optimization: bail early if file does not exist
if stat_cache is None and not os.path.exists(path):
raise MissingFileError(path)
headers = Headers([])
self.add_mime_headers(headers, path, url)
self.add_cache_headers(headers, path, url)
if self.allow_all_origins:
headers['Access-Control-Allow-Origin'] = '*'
if self.add_headers_function:
self.add_headers_function(headers, path, url)
return StaticFile(
path, headers.items(),
stat_cache=stat_cache,
encodings={
'gzip': path + '.gz', 'br': path + '.br'})
开发者ID:evansd,项目名称:whitenoise,代码行数:16,代码来源:base.py
示例14: _parse_headers
def _parse_headers(environ):
"""
Parse the environmental variables, looking for HTTP request headers.
:param environ: environmental variables
:type environ: dict
:return: request headers
:rtype: dict
"""
headers = Headers([])
for key, value in environ.items():
match = _HTTP_HEADER_REGEX.match(key)
if match is None:
continue
name = _normalize_header_name(match.group(0))
headers.add_header(name, value)
return headers
开发者ID:jlconnor,项目名称:wsgistack,代码行数:16,代码来源:request.py
示例15: __init__
def __init__(self, path, is_immutable, guess_type=mimetypes.guess_type, **config):
self.path = path
stat = os.stat(path)
self.mtime_tuple = gmtime(stat.st_mtime)
mimetype, encoding = guess_type(path)
mimetype = mimetype or 'application/octet-stream'
charset = self.get_charset(mimetype)
params = {'charset': charset} if charset else {}
self.headers = Headers([
('Last-Modified', formatdate(stat.st_mtime, usegmt=True)),
('Content-Length', str(stat.st_size)),
])
self.headers.add_header('Content-Type', str(mimetype), **params)
if encoding:
self.headers['Content-Encoding'] = encoding
max_age = self.FOREVER if is_immutable else config['max_age']
if max_age is not None:
self.headers['Cache-Control'] = 'public, max-age=%s' % max_age
if config['allow_all_origins']:
self.headers['Access-Control-Allow-Origin'] = '*'
gzip_path = path + self.GZIP_SUFFIX
if os.path.isfile(gzip_path):
self.gzip_path = gzip_path
self.headers['Vary'] = 'Accept-Encoding'
# Copy the headers and add the appropriate encoding and length
self.gzip_headers = Headers(self.headers.items())
self.gzip_headers['Content-Encoding'] = 'gzip'
self.gzip_headers['Content-Length'] = str(os.stat(gzip_path).st_size)
else:
self.gzip_path = self.gzip_headers = None
开发者ID:ionelmc,项目名称:whitenoise,代码行数:33,代码来源:base.py
示例16: newapp
def newapp(environ, start_response):
body = app(environ, capture)
status = resp['status']
headers = Headers(resp['headers'])
already = 'Content-Encoding' in headers
accepted = 'gzip' in environ.get('HTTP_ACCEPT_ENCODING', '')
if not accepted or already:
# no compress
start_response(status, list(headers.items()))
return body
content = gzip.compress(b''.join(body))
if hasattr(body, 'close'):
body.close()
headers['Content-Encoding'] = 'gzip'
start_response(status, list(headers.items()))
return [content]
开发者ID:disunbow,项目名称:saku,代码行数:18,代码来源:middleware.py
示例17: __init__
def __init__(self, body=b'', status=None, headers=None):
self.headers = Headers()
self._body = body
self._status_code = status or self.default_status
self._cookies = SimpleCookie()
if headers:
for name, value in headers.items():
self.headers.add_header(name, value)
开发者ID:c-bata,项目名称:kobin,代码行数:9,代码来源:environs.py
示例18: board_app
def board_app(env, resp):
path = env['PATH_INFO']
m = board_re.match(path)
board = m.group(1)
message = gateway.search_message(env.get('HTTP_ACCEPT_LANGUAGE', 'ja'))
headers = Headers([('Content-Type', 'text/html; charset=Shift_JIS')])
resp("200 OK", headers.items())
html = [
'<!DOCTYPE html>',
'<html><head>',
'<meta http-equiv="content-type" content="text/html; charset=Shift_JIS">',
'<title>%s - %s</title>' % (message['logo'], message['description']),
'<meta name="description" content="%s - %s">' % (message['logo'], message['description']),
'</head><body>',
'<h1>%s - %s</h1>' % (message['logo'], message['description']),
'</body></html>',
]
return ((c + '\n').encode('sjis', 'ignore') for c in html)
开发者ID:perillaseed,项目名称:saku,代码行数:20,代码来源:datd.py
示例19: finish_header
def finish_header(self):
self.file = BytesIO()
self.headers = Headers(self.headerlist)
cdis = self.headers.get('Content-Disposition', '')
ctype = self.headers.get('Content-Type', '')
if not cdis:
raise MultipartError('Content-Disposition header is missing.')
self.disposition, self.options = parse_options_header(cdis)
self.name = self.options.get('name')
self.filename = self.options.get('filename')
self.content_type, options = parse_options_header(ctype)
self.charset = options.get('charset') or self.charset
self.content_length = int(self.headers.get('Content-Length', '-1'))
开发者ID:hirunatan,项目名称:anillo,代码行数:13,代码来源:multipart.py
示例20: finish_header
def finish_header(self):
self.file = BytesIO()
self.headers = Headers(self.headerlist)
cdis = self.headers.get("Content-Disposition", "")
ctype = self.headers.get("Content-Type", "")
clen = self.headers.get("Content-Length", "-1")
if not cdis:
raise MultipartError("Content-Disposition header is missing.")
self.disposition, self.options = parse_options_header(cdis)
self.name = self.options.get("name")
self.filename = self.options.get("filename")
self.content_type, options = parse_options_header(ctype)
self.charset = options.get("charset") or self.charset
self.content_length = int(self.headers.get("Content-Length", "-1"))
开发者ID:cyberj,项目名称:pulsar,代码行数:14,代码来源:multipart.py
注:本文中的wsgiref.headers.Headers类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论