python类Response()的实例源码

wsgi_test.py 文件源码 项目:serverless-wsgi 作者: logandk 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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)
endpoints.py 文件源码 项目:talisker 作者: canonical-ols 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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
endpoints.py 文件源码 项目:talisker 作者: canonical-ols 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:Sci-Finder 作者: snverse 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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)
utils.py 文件源码 项目:spichi 作者: ifkite 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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()
shared.py 文件源码 项目:oclubs 作者: SHSIDers 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:RPoint 作者: george17-meet 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
create.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
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
create.py 文件源码 项目:deb-python-falcon 作者: openstack 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
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
fixtures.py 文件源码 项目:apm-agent-python 作者: elastic 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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)
wrappers.py 文件源码 项目:oa_qian 作者: sunqb 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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)
wrappers.py 文件源码 项目:oa_qian 作者: sunqb 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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))
wrappers.py 文件源码 项目:oa_qian 作者: sunqb 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
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)
wrappers.py 文件源码 项目:oa_qian 作者: sunqb 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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'
__init__.py 文件源码 项目:Isa-Framework 作者: c0hb1rd 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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
__init__.py 文件源码 项目:Isa-Framework 作者: c0hb1rd 项目源码 文件源码 阅读 30 收藏 0 点赞 0 评论 0
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

    # ????
__init__.py 文件源码 项目:Isa-Framework 作者: c0hb1rd 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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)


# ?????????????????
exceptions.py 文件源码 项目:Indushell 作者: SecarmaLabs 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:Liljimbo-Chatbot 作者: chrisjim316 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:flask_system 作者: prashasy 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:ShelbySearch 作者: Agentscreech 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:FileStoreGAE 作者: liantian-cn 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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)
test_host_http_server.py 文件源码 项目:py 作者: stencila 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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()
exceptions.py 文件源码 项目:python-group-proj 作者: Sharcee 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
exceptions.py 文件源码 项目:islam-buddy 作者: hamir 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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)
test.py 文件源码 项目:nirum-python 作者: spoqa 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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)
base_handler.py 文件源码 项目:territoriali-backend 作者: algorithm-ninja 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
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)
base_handler.py 文件源码 项目:territoriali-backend 作者: algorithm-ninja 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
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
exceptions.py 文件源码 项目:jieba-GAE 作者: liantian-cn 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
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)
__init__.py 文件源码 项目:micromaterials-api 作者: lpmi-13 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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


问题


面经


文章

微信
公众号

扫码关注公众号