我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用flask.Request()。
def create_from_request(cls: t.Type['LTI'], req: flask.Request) -> 'LTI': params = req.form.copy() lti_provider = models.LTIProvider.query.filter_by( key=params['oauth_consumer_key'] ).first() if lti_provider is None: lti_provider = models.LTIProvider(key=params['oauth_consumer_key']) db.session.add(lti_provider) db.session.commit() params['lti_provider_id'] = lti_provider.id # This is semi sensitive information so it should not end up in the JWT # token. launch_params = {} for key, value in params.items(): if not key.startswith('oauth'): launch_params[key] = value self = cls(launch_params, lti_provider) auth.ensure_valid_oauth(self.key, self.secret, req) return self
def setUp(self) -> None: """ Set up some mock model classes and create an endpoint """ self.session = mock.MagicMock(spec=Session) # type: Session self.request = mock.MagicMock(spec=Request) # type: Request self.service_list = mock.MagicMock( spec=ServiceList ) # type: ServiceList self._app = Flask(__name__) self._app.add_url_rule( '/', view_func=ServiceDetail.as_view( ServiceDetail.__name__, self.session ) ) self._context = self._app.test_request_context() self._context.push()
def test_invalid_post(self, request_body: dict) -> None: assume("name" not in request_body.keys()) assume("description" not in request_body.keys()) assume("job_registration_schema" not in request_body.keys()) assume("job_result_schema" not in request_body.keys()) request = mock.MagicMock(spec=Request) request.get_json = mock.MagicMock(return_value=request_body) endpoint = ServicesList( self.session, request ) with self.assertRaises(endpoint.Abort): endpoint.post() self.assertTrue(endpoint.errors)
def __init__( self, session: Session, flask_request: Request=request, job_list_model: Optional[JobListInterface]=None ) -> None: """ :param session: The session to use :param flask_request: The Flask request that this endpoint needs to process """ super(self.__class__, self).__init__(session, flask_request) if job_list_model is None: self.job_list = JobListModel(self.database_session) else: self.job_list = job_list_model
def __init__( self, session: Session, flask_request: Request=request, metadata: APIMetadataModelInterface=MetadataModel() ) -> None: """ :param session: The SQLAlchemy ORM Session to use for communicating with the database. This is required by ``AbstractEndpoint`` in order to perform its transaction management :param flask_request: The request to process. By default, this is flask's ``request`` variable :param metadata: The metadata to serialize. By default, this is an instance of the ``MetadataModel`` that pulls all of its values from the ``config`` script. """ super(APIMetadata, self).__init__(session, request=flask_request) self._api_metadata = metadata
def __init__( self, session: Session, flask_request: Request=request, service_list: Optional[ServiceListInterface]=None ) -> None: """ Create a service list model that is capable of getting services from the DB :param session: The database session to use """ super(self.__class__, self).__init__(session, flask_request) if service_list is not None: self.service_list = service_list else: self.service_list = ServiceListModel(session)
def will_handle_ui(self, request): """ Called by OctoPrint to determine if the mixin implementation will be able to handle the ``request`` provided as a parameter. Return ``True`` here to signal that your implementation will handle the request and that the result of its :meth:`~octoprint.plugin.UiPlugin.on_ui_render` method is what should be served to the user. The execution order of calls to this method can be influenced via the sorting context ``UiPlugin.will_handle_ui``. Arguments: request (flask.Request): A Flask `Request <http://flask.pocoo.org/docs/0.10/api/#flask.Request>`_ object. Returns: bool: ``True`` if the the implementation will serve the request, ``False`` otherwise. """ return False
def test_modified_url_encoding(self): class ModifiedRequest(flask.Request): url_charset = 'euc-kr' app = flask.Flask(__name__) app.testing = True app.request_class = ModifiedRequest app.url_map.charset = 'euc-kr' @app.route('/') def index(): return flask.request.args['foo'] rv = app.test_client().get(u'/?foo=????'.encode('euc-kr')) self.assert_equal(rv.status_code, 200) self.assert_equal(rv.data, u'????'.encode('utf-8'))
def process_xml(self, xml: str) -> None: ''' Parse Outcome Request data from XML. ''' root = t.cast(t.Mapping, objectify.fromstring(xml)) self.message_identifier = str( root['imsx_POXHeader']['imsx_POXRequestHeaderInfo'] ['imsx_messageIdentifier'] ) try: result = root['imsx_POXBody']['replaceResultRequest'] self.operation = REPLACE_REQUEST # Get result sourced id from resultRecord self.lis_result_sourcedid = result.resultRecord.\ sourcedGUID.sourcedId self.score = str(result.resultRecord.result.resultScore.textString) except (KeyError, TypeError, AttributeError): pass try: result = root['imsx_POXBody']['deleteResultRequest'] self.operation = DELETE_REQUEST # Get result sourced id from resultRecord self.lis_result_sourcedid = result['resultRecord']['sourcedGUID' ]['sourcedId'] except (KeyError, TypeError): pass try: result = root['imsx_POXBody']['readResultRequest'] self.operation = READ_REQUEST # Get result sourced id from resultRecord self.lis_result_sourcedid = result['resultRecord']['sourcedGUID' ]['sourcedId'] except (KeyError, TypeError): pass
def setUp(self) -> None: TestAPI.setUp(self) self.session = mock.MagicMock(spec=Session) self.request = mock.MagicMock(spec=Request) self.service_list = mock.MagicMock(spec=ServiceList) self.testing_app.add_url_rule( '/test_url/<service_id>', view_func=JobsForServiceEndpoint.as_view( JobsForServiceEndpoint.__name__ ) )
def setUp(self) -> None: self.session = mock.MagicMock(spec=Session) self.request = mock.MagicMock(spec=Request) app = Flask(__name__) app.add_url_rule( '/', view_func=JSONSchemaValidator.as_view( JSONSchemaValidator.__name__ ) ) self.context = app.test_request_context() self.context.push()
def setUp(self) -> None: """ Set up a mock session and push a testing application context for Flask """ self.session = mock.MagicMock(spec=Session) # type: Session self.request = mock.MagicMock(spec=Request) # type: Request self.testing_app = TestingWSGIAPPFactory().app self._app_context = self.testing_app.test_request_context() self._app_context.push()
def setUp(self) -> None: """ Set up the test """ self.session = mock.MagicMock(spec=Session) self.request = mock.MagicMock(spec=Request) app = Flask(__name__) app.add_url_rule( '/', view_func=JobDetail.as_view( JobDetail.__name__, ) ) self.context = app.test_request_context() self.context.push()
def setUp(self) -> None: """ Create a fake service list endpoint """ TestAPI.setUp(self) self.session = mock.MagicMock(spec=Session) # type: Session self.request = mock.MagicMock(spec=Request) # type: Request
def test_valid_post(self, request_body: dict): request = mock.MagicMock(spec=Request) request.get_json = mock.MagicMock(return_value=request_body) endpoint = ServicesList( self.session, request ) response = endpoint.post() self.assertEqual(response.status_code, 201) self.assertIn('Location', response.headers)
def setUp(self) -> None: """ Create a mock Flask request a mock DB session, and have the endpoint handle these fake requests """ self.request = mock.MagicMock(spec=Request) self.session = mock.MagicMock(spec=Session) self._app = Flask(__name__) self._app.add_url_rule('/', view_func=JobsList.as_view( JobsList.__name__ )) self._context = self._app.test_request_context() self._context.push()
def setUp(self) -> None: self.session = mock.MagicMock(spec=Session) # type: Session self.request = mock.MagicMock(spec=Request) # type: Request app = Flask(__name__) app.add_url_rule('/', view_func=NextJob.as_view( NextJob.__name__ )) self.context = app.test_request_context() self.context.push()
def __init__( self, session: Session, flask_request: Request=request, service_list: Optional[ServiceList]=None, validator_factory: Optional[type]=None ): super(JobsForServiceEndpoint, self).__init__( session, flask_request, service_list=service_list ) if validator_factory is None: self._validator_factory = JSONSchemaValidator else: self._validator_factory = validator_factory
def __init__( self, session: Session, flask_request: Request=request, job_list:Optional[JobList]=None, validator_factory: Type[JsonschemaValidator]=JsonschemaValidator ) -> None: super(JobDetail, self).__init__( session, flask_request, job_list ) self._validator_factory = validator_factory
def __init__( self, session: Session, request: Request=flask_request ) -> None: """ :param session: The database session to be used for interacting with the database. :param request: The Flask request object to use for running the request """ super(AbstractEndpoint, self).__init__() self._session = session self._errors = [] self._request = request
def __init__( self, session: Session, flask_request: Request=request, job_list: Optional[JobList]=None ): super(AbstractEndpointForJob, self).__init__(session, flask_request) if job_list is None: self._job_list = JobListModel(self.database_session)
def configure(self, binder): binder.bind_scope(self.request_scope_class) binder.bind(flask.Flask, to=self.app, scope=singleton) binder.bind(Config, to=self.app.config, scope=singleton) binder.bind(Request, to=lambda: flask.request)
def stub_request(query_parameters=''): return Request({'QUERY_STRING': query_parameters})
def on_api_get(self, request): """ Called by OctoPrint upon a GET request to ``/api/plugin/<plugin identifier>``. ``request`` will contain the received `Flask request object <http://flask.pocoo.org/docs/0.9/api/#flask.Request>`_ which you may evaluate for additional arguments supplied with the request. If your plugin returns nothing here, OctoPrint will return an empty response with return code ``204 No content`` for you. You may also return regular responses as you would return from any Flask view here though, e.g. ``return flask.jsonify(result="some json result")`` or ``return flask.make_response("Not found", 404)``. :param request: the Flask request object :return: ``None`` in which case OctoPrint will generate a ``204 No content`` response with empty body, or optionally a proper Flask response. """ return None
def __init__(self, environ, *args, **kwargs): # apply environment wrapper to provided WSGI environment flask.Request.__init__(self, self.environment_wrapper(environ), *args, **kwargs)
def cookies(self): # strip cookie_suffix from all cookies in the request, return result cookies = flask.Request.cookies.__get__(self) result = dict() desuffixed = dict() for key, value in cookies.items(): if key.endswith(self.cookie_suffix): desuffixed[key[:-len(self.cookie_suffix)]] = value else: result[key] = value result.update(desuffixed) return result
def cookie_suffix(self): """ Request specific suffix for set and read cookies We need this because cookies are not port-specific and we don't want to overwrite our session and other cookies from one OctoPrint instance on our machine with those of another one who happens to listen on the same address albeit a different port. """ return "_P" + self.server_port