def __call__(self, environ, start_response):
self.last_environ = environ
if self.enable_binary:
mimetype = 'image/jpeg'
else:
mimetype = 'text/plain'
response = Response(u'Hello World ?!', mimetype=mimetype)
cookies = [
('CUSTOMER', 'WILE_E_COYOTE'),
('PART_NUMBER', 'ROCKET_LAUNCHER_0002'),
('LOT_NUMBER', '42')
]
for cookie in cookies[:self.cookie_count]:
response.set_cookie(cookie[0], cookie[1])
print("application debug #1", file=environ['wsgi.errors'])
return response(environ, start_response)
python类Response()的实例源码
def private(f):
"""Only allow approved source addresses."""
@functools.wraps(f)
def wrapper(self, request):
if not request.access_route:
# this means something probably bugged in werkzeug, but let's fail
# gracefully
return Response('no client ip provided', status='403')
ip_str = request.access_route[-1]
if isinstance(ip_str, bytes):
ip_str = ip_str.decode('utf8')
ip = ip_address(ip_str)
if ip.is_loopback or any(ip in network for network in get_networks()):
return f(self, request)
else:
msg = PRIVATE_BODY_RESPONSE_TEMPLATE.format(
ip_str,
force_unicode(request.remote_addr),
request.headers.get('x-forwarded-for'))
return Response(msg, status='403')
return wrapper
def __call__(self, environ, start_response):
request = Request(environ)
if request.path.startswith(self.prefix):
method = request.path[len(self.prefix):]
if method == '':
# no trailing /
start_response('302', [('location', self.prefix + '/')])
return ''
try:
funcname = self.urlmap[method]
func = getattr(self, funcname)
except (KeyError, AttributeError):
response = Response(status=404)
else:
response = func(request)
return response(environ, start_response)
else:
return self.app(environ, start_response)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def redirect(location, status=302):
"""
which is seldom used in api server
"""
from werkzeug.wrappers import Response
from werkzeug.urls import iri_to_uri
location = iri_to_uri(location, safe_conversion=True)
return Response(
"<!DOCTYPE html>\
<html>\
<h1>Redirecting...</h1>\
<a href='{0}'>touch this to make manually redirection</a>\
</html>"
.format(location), status=status, headers={'Location': location})
# sample
# app = create_app()
# @app.route('/', 'home')
# def home(request):
# app.UploadHandlerClass(request).save()
# OR give some specific filenames
# app.UploadHandlerClass(request, ['image']).save()
def download_xlsx(filename, info):
'''Çreate xlsx file for given info and download it '''
output = StringIO()
workbook = xlsxwriter.Workbook(output, {'in_memory': True})
worksheet = workbook.add_worksheet()
row_num = 0
col_num = 0
for row in info:
for grid in row:
worksheet.write(row_num, col_num, grid)
col_num += 1
col_num = 0
row_num += 1
workbook.close()
output.seek(0)
headers = Headers()
headers.set('Content-Disposition', 'attachment', filename=filename)
return Response(output.read(), mimetype='application/vnd.openxmlformats-'
'officedocument.spreadsheetml.sheet', headers=headers)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def flask(body, headers):
import flask
path = '/hello/<account_id>/test'
flask_app = flask.Flask('hello')
@flask_app.route(path)
def hello(account_id):
request = flask.request
user_agent = request.headers['User-Agent'] # NOQA
limit = request.args.get('limit', '10') # NOQA
return flask.Response(body, headers=headers,
mimetype='text/plain')
return flask_app
def werkzeug(body, headers):
import werkzeug.wrappers as werkzeug
from werkzeug.routing import Map, Rule
path = '/hello/<account_id>/test'
url_map = Map([Rule(path, endpoint='hello')])
@werkzeug.Request.application
def hello(request):
user_agent = request.headers['User-Agent'] # NOQA
limit = request.args.get('limit', '10') # NOQA
adapter = url_map.bind_to_environ(request.environ) # NOQA
endpoint, values = adapter.match() # NOQA
aid = values['account_id'] # NOQA
return werkzeug.Response(body, headers=headers,
mimetype='text/plain')
return hello
def __call__(self, environ, start_response):
code = self.code
content = self.content
request = Request(environ)
self.requests.append(request)
data = request.data
if request.content_encoding == 'deflate':
data = zlib.decompress(data)
data = data.decode(request.charset)
if request.content_type == 'application/json':
data = json.loads(data)
self.payloads.append(data)
validator = VALIDATORS.get(request.path, None)
if validator and not self.skip_validate:
try:
validator.validate(data)
code = 202
except jsonschema.ValidationError as e:
code = 400
content = json.dumps({'status': 'error', 'message': str(e)})
response = Response(status=code)
response.headers.clear()
response.headers.extend(self.headers)
response.data = content
return response(environ, start_response)
def test_type_forcing(self):
def wsgi_application(environ, start_response):
start_response('200 OK', [('Content-Type', 'text/html')])
return ['Hello World!']
base_response = wrappers.BaseResponse('Hello World!', content_type='text/html')
class SpecialResponse(wrappers.Response):
def foo(self):
return 42
# good enough for this simple application, but don't ever use that in
# real world examples!
fake_env = {}
for orig_resp in wsgi_application, base_response:
response = SpecialResponse.force_type(orig_resp, fake_env)
assert response.__class__ is SpecialResponse
self.assert_strict_equal(response.foo(), 42)
self.assert_strict_equal(response.get_data(), b'Hello World!')
self.assert_equal(response.content_type, 'text/html')
# without env, no arbitrary conversion
self.assert_raises(TypeError, SpecialResponse.force_type, wsgi_application)
def test_etag_response_mixin_freezing(self):
class WithFreeze(wrappers.ETagResponseMixin, wrappers.BaseResponse):
pass
class WithoutFreeze(wrappers.BaseResponse, wrappers.ETagResponseMixin):
pass
response = WithFreeze('Hello World')
response.freeze()
self.assert_strict_equal(response.get_etag(),
(text_type(wrappers.generate_etag(b'Hello World')), False))
response = WithoutFreeze('Hello World')
response.freeze()
self.assert_equal(response.get_etag(), (None, None))
response = wrappers.Response('Hello World')
response.freeze()
self.assert_equal(response.get_etag(), (None, None))
def test_ranges(self):
# basic range stuff
req = wrappers.Request.from_values()
assert req.range is None
req = wrappers.Request.from_values(headers={'Range': 'bytes=0-499'})
self.assert_equal(req.range.ranges, [(0, 500)])
resp = wrappers.Response()
resp.content_range = req.range.make_content_range(1000)
self.assert_equal(resp.content_range.units, 'bytes')
self.assert_equal(resp.content_range.start, 0)
self.assert_equal(resp.content_range.stop, 500)
self.assert_equal(resp.content_range.length, 1000)
self.assert_equal(resp.headers['Content-Range'], 'bytes 0-499/1000')
resp.content_range.unset()
assert 'Content-Range' not in resp.headers
resp.headers['Content-Range'] = 'bytes 0-499/1000'
self.assert_equal(resp.content_range.units, 'bytes')
self.assert_equal(resp.content_range.start, 0)
self.assert_equal(resp.content_range.stop, 500)
self.assert_equal(resp.content_range.length, 1000)
def test_dynamic_charset_response_mixin(self):
class MyResponse(wrappers.DynamicCharsetResponseMixin, Response):
default_charset = 'utf-7'
resp = MyResponse(mimetype='text/html')
assert resp.charset == 'utf-7'
resp.charset = 'utf-8'
assert resp.charset == 'utf-8'
assert resp.mimetype == 'text/html'
assert resp.mimetype_params == {'charset': 'utf-8'}
resp.mimetype_params['charset'] = 'iso-8859-15'
assert resp.charset == 'iso-8859-15'
resp.set_data(u'Hällo Wörld')
assert b''.join(resp.iter_encoded()) == \
u'Hällo Wörld'.encode('iso-8859-15')
del resp.headers['content-type']
try:
resp.charset = 'utf-8'
except TypeError as e:
pass
else:
assert False, 'expected type error on charset setting without ct'
def capture(f):
def decorator(*args, **options):
# ??????
try:
# ??????
rep = f(*args, **options)
except SYLFkException as e:
# ???? SYLFkException ??????????????????????????? ERROR_MAP ????????????????
if e.code in ERROR_MAP and ERROR_MAP[e.code]:
# ?????????
rep = ERROR_MAP[e.code]
# ???????? 100??????????? 500 ?????
status = int(e.code) if int(e.code) >= 100 else 500
# ????????????????????????????????????????????
return rep if isinstance(rep, Response) or rep is None else Response(rep(), content_type=content_type, status=status)
else:
# ?????????????
raise e
# ???????????
return rep
# ?????
return decorator
def dispatch_static(self, static_path):
# ????????
key = parse_static_key(static_path)
# ????????????????????????????????
if os.path.exists(static_path):
doc_type = TYPE_MAP.get(key, 'text/plain')
with open(static_path, 'rb') as f:
rep = f.read()
# ????????
return Response(rep, content_type=doc_type)
else:
# ?????????
raise exceptions.PageNotFoundError
# ????
def render_json(data):
# ????????????
content_type = "text/plain"
# ??? Dict ?? List ????????? JSON ????
if isinstance(data, dict) or isinstance(data, list):
# ? data ??? JSON ????
data = json.dumps(data)
# ??????? JSON ??
content_type = "application/json"
# ?????????
return Response(data, content_type="%s; charset=UTF-8" % content_type, status=200)
# ?????????????????
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def test_handle():
s = HostHttpServer(host)
c = Client(s, Response)
s.start()
r = c.get('/')
assert r.status == '403 FORBIDDEN'
r = c.get('/?ticket=%s' % s.ticket_create())
assert r.status == '200 OK'
r = c.get('/foo/bar/?ticket=%s' % s.ticket_create())
assert r.status == '500 INTERNAL SERVER ERROR'
s.stop()
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def __init__(self, url, service_impl_cls):
self.url = url_endswith_slash(url)
self.scheme, self.host, self.path, _, _ = urllib.parse.urlsplit(
self.url
)
def null_app(environ, start_response):
start_response('404 Not Found', [('Content-Type', 'text/plain')])
return ['Not Found']
wsgi_app = WsgiApp(service_impl_cls())
if self.path != '/':
self.wsgi_app = DispatcherMiddleware(
null_app, {self.path[:-1]: wsgi_app}
)
else:
self.wsgi_app = wsgi_app
self.wsgi_test_client = Client(self.wsgi_app, Response)
def raise_exc(cls, code, message):
"""
Raise an HTTPException with a code and a message sent in a json like
{
"code": code
"message": message
}
:param cls: HTTPException of the error, for example NotFound, BadRequest, NotAuthorized
:param code: A brief message for the exception, like MISSING_PARAMETER
:param message: A longer description of the error
:return: Nothing, raise the provided exception with the correct response
"""
response = Response()
response.mimetype = "application/json"
response.status_code = cls.code
response.data = json.dumps({
"code": code,
"message": message
})
Logger.warning(cls.__name__.upper(), code + ": " + message)
raise cls(response=response)
def handle(self, endpoint, route_args, request):
"""
Handle a request in the derived handler. The request is routed to the correct method using *endpoint*
:param endpoint: A string with the name of the class method to call with (route_args, request) as parameters,
this method should return a Response or call self.raise_exc. *NOTE*: the method MUST be implemented in the
derived class
:param route_args: The route parameters, the parameters extracted from the matching route in the URL
:param request: The Request object, request.args contains the query parameters of the request
:return: Return a Response if the request is successful, an HTTPException if an error occurred
"""
try:
data = BaseHandler._call(self.__getattribute__(endpoint), route_args, request)
response = Response()
if data is not None:
response.code = 200
response.mimetype = "application/json"
response.data = json.dumps(data)
else:
response.code = 204
return response
except HTTPException as e:
return e
def abort(status, *args, **kwargs):
'''
Raises an :py:exc:`HTTPException` for the given status code or WSGI
application::
abort(404) # 404 Not Found
abort(Response('Hello World'))
Can be passed a WSGI application or a status code. If a status code is
given it's looked up in the list of exceptions and will raise that
exception, if passed a WSGI application it will wrap it in a proxy WSGI
exception and raise that::
abort(404)
abort(Response('Hello World'))
'''
return _aborter(status, *args, **kwargs)
def respond(response_data, *args, **kwargs):
"""
Create a C{Response}.
@param response_data: metadata about the response such as the view to be
rendered.
@param args: other positional arguments.
@param kwargs: other named arguments.
@return: a C{Response}; never C{None}.
"""
if isinstance(response_data, Response):
response = response_data
else:
(view_name, model, status_code) = deconstruct(response_data)
responsifier = current_app.responsifier
response = responsifier.responsify(model,
status_code=status_code,
view_name=view_name,
*args, **kwargs)
response.status_code = status_code
return response
