本文整理汇总了Python中flask.request.get_data方法的典型用法代码示例。如果您正苦于以下问题：Python request.get_data方法的具体用法？Python request.get_data怎么用？Python request.get_data使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块flask.request的用法示例。

在下文中一共展示了request.get_data方法的28个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: do_POST

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def do_POST(self, service=""):

log.debug("NebHookServer: Plugin=%s : Incoming request from %s",

service, request.remote_addr)

if service.split("/")[0] not in self.plugin_mappings:

return ("", 404, {})

plugin = self.plugin_mappings[service.split("/")[0]]

try:

# tuple (body, status_code, headers)

response = plugin.on_receive_webhook(

request.url,

request.get_data(),

request.remote_addr,

request.headers

)

if response:

return response

return ("", 200, {})

except Exception as e:

log.exception(e)

return ("", 500, {})

开发者ID:matrix-org，项目名称:Matrix-NEB，代码行数:24，

示例2: predict_image_handler

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def predict_image_handler():

try:

imageData = None

if ('imageData' in request.files):

imageData = request.files['imageData']

else:

imageData = io.BytesIO(request.get_data())

#img = scipy.misc.imread(imageData)

img = Image.open(imageData)

results = predict_image(img)

return json.dumps(results)

except Exception as e:

print('EXCEPTION:', str(e))

return 'Error processing image', 500

# Like the CustomVision.ai Prediction service /url route handles url's

# in the body of hte request of the form:

# { 'Url': ''}

开发者ID:jamesbannan，项目名称:pluralsight，代码行数:22，

示例3: predict_image_handler

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def predict_image_handler(project=None, publishedName=None):

try:

imageData = None

if ('imageData' in request.files):

imageData = request.files['imageData']

elif ('imageData' in request.form):

imageData = request.form['imageData']

else:

imageData = io.BytesIO(request.get_data())

img = Image.open(imageData)

results = predict_image(img)

return jsonify(results)

except Exception as e:

print('EXCEPTION:', str(e))

return 'Error processing image', 500

# Like the CustomVision.ai Prediction service /url route handles url's

# in the body of hte request of the form:

# { 'Url': ''}

开发者ID:Azure-Samples，项目名称:Custom-vision-service-iot-edge-raspberry-pi，代码行数:23，

示例4: slack_app

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def slack_app():

if not signature_verifier.is_valid_request(request.get_data(), request.headers):

return make_response("invalid request", 403)

if "command" in request.form \

and request.form["command"] == "/view":

trigger_id = request.form["trigger_id"]

return open_modal(trigger_id)

elif "payload" in request.form:

payload = json.loads(request.form["payload"])

if payload["type"] == "view_submission" \

and payload["view"]["callback_id"] == "modal-id":

submitted_data = payload["view"]["state"]["values"]

print(submitted_data) # {'b-id': {'a-id': {'type': 'plain_text_input', 'value': 'your input'}}}

return make_response("", 200)

if payload["type"] == "shortcut" \

and payload["callback_id"] == "view-test":

return open_modal(payload["trigger_id"])

return make_response("", 404)

开发者ID:slackapi，项目名称:python-slackclient，代码行数:23，

示例5: post

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def post(self):

parameters = json.loads(request.get_data(as_text=True))

partial_element = parameters['partial_element']

if len(partial_element) < 2:

return flask.jsonify({'success': True, 'result': []})

try:

interactions = cellphonedb_app.cellphonedb.query.autocomplete_launcher(partial_element)

response = {

'success': True,

'result': interactions.to_dict(orient='records')

}

except:

response = {

'success': False

}

print(traceback.print_exc(file=sys.stdout))

return flask.jsonify(response)

开发者ID:Teichlab，项目名称:cellphonedb，代码行数:23，

示例6: post

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def post(self):

parameters = json.loads(request.get_data(as_text=True))

complex_name = parameters['complex_name']

deconvoluted = cellphonedb_app.cellphonedb.query.get_complex_deconvoluted(complex_name)

if deconvoluted.empty:

self.attach_error(

{'code': 'element_not_found', 'title': '%s is not CellPhoneDB Complex' % complex_name,

'details': '%s is not present in CellPhoneDB complex table' % complex_name})

else:

self._attach_csv(deconvoluted.to_csv(index=False, sep=','), 'result')

self._commit_attachments()

return Response(self._msg.as_string(), mimetype='multipart/form-data; boundary="%s"' % self._msg.get_boundary())

开发者ID:Teichlab，项目名称:cellphonedb，代码行数:19，

示例7: post

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def post(self):

parameters = json.loads(request.get_data(as_text=True))

receptor = parameters['receptor']

interactions = cellphonedb_app.cellphonedb.query.find_interactions_by_element(receptor)

if interactions.empty:

self.attach_error(

{'code': 'result_not_found', 'title': '%s is not CellPhoneDB interactor' % receptor,

'details': '%s is not present in CellPhoneDB interactor enabled table' % receptor})

else:

self._attach_csv(interactions.to_csv(index=False, sep=','), 'ligands')

self._commit_attachments()

return Response(self._msg.as_string(), mimetype='multipart/form-data; boundary="%s"' % self._msg.get_boundary())

示例8: post

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def post(self):

id = request.get_data('id')

project = id.decode('utf-8')

if not project:

return jsonify({'msg': u'没有发送数据', 'code': 38, 'data': ''})

project_is = Project.query.filter_by(project_name=project).first()

if not project_is:

return jsonify(({'msg': u'成功', 'code': 200, 'data': []}))

testreport = TestResult.query.filter_by(projects_id=project_is.id, status=False).order_by('-id').all()

testreportlist = []

for test in testreport:

testreportlist.append({'test_num': test.test_num, 'pass_num': test.pass_num,

'fail_num': test.fail_num, 'hour_time': str(test.hour_time),

'test_rep': test.test_rep, 'test_log': test.test_log,

'Exception_num': test.Exception_num, 'can_num': test.can_num,

'wei_num': test.wei_num, 'test_time': str(test.test_time),

'Test_user_id': test.users.username, 'id': test.id,

'fenshu': test.pass_num / test.test_num})

return jsonify(({'msg': u'成功', 'code': 200, 'data': (testreportlist)}))

开发者ID:liwanlei，项目名称:FXTest，代码行数:21，

示例9: test_flask_large_json_request

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def test_flask_large_json_request(sentry_init, capture_events, app):

sentry_init(integrations=[flask_sentry.FlaskIntegration()])

data = {"foo": {"bar": "a" * 2000}}

@app.route("/", methods=["POST"])

def index():

assert request.get_json() == data

assert request.get_data() == json.dumps(data).encode("ascii")

assert not request.form

capture_message("hi")

return "ok"

events = capture_events()

client = app.test_client()

response = client.post("/", content_type="application/json", data=json.dumps(data))

assert response.status_code == 200

(event,) = events

assert event["_meta"]["request"]["data"]["foo"]["bar"] == {

"": {"len": 2000, "rem": [["!limit", "x", 509, 512]]}

}

assert len(event["request"]["data"]["foo"]["bar"]) == 512

开发者ID:getsentry，项目名称:sentry-python，代码行数:26，

示例10: test_flask_empty_json_request

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def test_flask_empty_json_request(sentry_init, capture_events, app, data):

sentry_init(integrations=[flask_sentry.FlaskIntegration()])

@app.route("/", methods=["POST"])

def index():

assert request.get_json() == data

assert request.get_data() == json.dumps(data).encode("ascii")

assert not request.form

capture_message("hi")

return "ok"

events = capture_events()

client = app.test_client()

response = client.post("/", content_type="application/json", data=json.dumps(data))

assert response.status_code == 200

(event,) = events

assert event["request"]["data"] == data

开发者ID:getsentry，项目名称:sentry-python，代码行数:21，

示例11: test_flask_medium_formdata_request

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def test_flask_medium_formdata_request(sentry_init, capture_events, app):

sentry_init(integrations=[flask_sentry.FlaskIntegration()])

data = {"foo": "a" * 2000}

@app.route("/", methods=["POST"])

def index():

assert request.form["foo"] == data["foo"]

assert not request.get_data()

assert not request.get_json()

capture_message("hi")

return "ok"

events = capture_events()

client = app.test_client()

response = client.post("/", data=data)

assert response.status_code == 200

(event,) = events

assert event["_meta"]["request"]["data"]["foo"] == {

"": {"len": 2000, "rem": [["!limit", "x", 509, 512]]}

}

assert len(event["request"]["data"]["foo"]) == 512

开发者ID:getsentry，项目名称:sentry-python，代码行数:26，

示例12: rename_path_on_json

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def rename_path_on_json():

req_data = json.loads(request.get_data() or '{}')

file_objs = req_data['file_objs']

path = req_data['path']

# handle usual error

if not os.path.exists(path):

return jsonify({'response': [{'file_name': f'{path} does not exist'}]}), 400

if not os.path.isdir(path):

return jsonify({'response': [{'file_name': f'{path} is not a valid directory for scan'}]}), 400

def long_process():

for each_file_process in EmbyFileStructure.rename_directory(path, file_objs):

yield json.dumps({'log': each_file_process})+'\n'

return Response(long_process(), mimetype='text/event-stream')

开发者ID:ddd354，项目名称:JAVOneStop，代码行数:18，

示例13: update_db_jav

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def update_db_jav():

req_data = json.loads(request.get_data() or '{}')

jav_pk = req_data.get('pk')

update_data = req_data.get('data')

if not jav_pk:

return jsonify({'error': 'no pk found in posted json'}), 400

db_conn = JavManagerDB()

try:

current_jav_obj = dict(db_conn.get_by_pk(jav_pk))

except DoesNotExist:

current_jav_obj = {'car': jav_pk}

current_jav_obj.update(update_data)

db_conn.upcreate_jav(current_jav_obj)

return jsonify({'success': dict(current_jav_obj)})

开发者ID:ddd354，项目名称:JAVOneStop，代码行数:20，

示例14: update_jav_dict

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def update_jav_dict():

# update db jav dict, also rewrite nfo and images

req_data = json.loads(request.get_data() or '{}')

update_dict = req_data['update_dict']

# update db

db_conn = JavManagerDB()

db_conn.upcreate_jav(update_dict)

file_writer = EmbyFileStructure(return_default_config_string('file_path'))

# file structure operations

try:

jav_obj = file_writer.create_folder_for_existing_jav(update_dict)

except KeyError as e:

_car = update_dict.get('car', 'Unknown')

update_dict.append(json.dumps({'log': f'error: {e}, skipping {_car}'}))

# write images

file_writer.write_images(jav_obj)

# write nfo

file_writer.write_nfo(jav_obj)

# move video file

jav_obj = file_writer.move_existing_file(jav_obj)

return jsonify({'success': jav_obj}) # post updated jav_obj back to UI

开发者ID:ddd354，项目名称:JAVOneStop，代码行数:27，

示例15: restructure_jav

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def restructure_jav():

req_data = json.loads(request.get_data() or '{}')

update_dict = req_data['update_dict']

file_writer = EmbyFileStructure(return_default_config_string('file_path'))

# file structure operations

try:

jav_obj = file_writer.create_folder_for_existing_jav(update_dict)

except KeyError as e:

_car = update_dict.get('car', 'Unknown')

update_dict.append(json.dumps({'log': f'error: {e}, skipping {_car}'}))

# write images

file_writer.write_images(jav_obj)

# write nfo

file_writer.write_nfo(jav_obj)

# move video file

jav_obj = file_writer.move_existing_file(jav_obj)

return jsonify({'success': 'good'})

开发者ID:ddd354，项目名称:JAVOneStop，代码行数:21，

示例16: handle

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def handle():

try:

url_data = request.get_data()

print(url_data)

'''Slacks interactive message request payload is in the form of

application/x-www-form-urlencoded JSON string. Getting first actions parameter

from it.'''

url_data = json.loads(parse_qs(url_data.decode('utf-8'))['payload'][0])['actions'][0]

eph_value = True if url_data['value'] == "yes" else False

print(url_data['name'] + " : " + url_data['value'] + " : " + str(eph_value))

if eph_value:

params = url_data['name'].split('|')

user_id = params[1]

channel_id = params[2]

file_id = params[3]

file_permalink = params[4]

comment = params[5]

timestamp = params[6]

i = pool.apply_async(download_file, [file_permalink, file_id, channel_id, comment, user_id, timestamp])

else:

print('---No chosen---')

except Exception as err:

print(err)

finally:

return jsonify({"response_type": "ephemeral", "replace_original": "true", "delete_original": "true", "text": ""})

开发者ID:dibyadas，项目名称:imagery，代码行数:27，

示例17: post_template_preview

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def post_template_preview(template_id):

# The payload is empty when there are no place holders in the template.

_data = request.get_data(as_text=True)

if not _data:

_data = {}

else:

_data = get_valid_json()

_data['id'] = template_id

data = validate(_data, post_template_preview_request)

template = templates_dao.dao_get_template_by_id_and_service_id(

template_id, authenticated_service.id)

template_object = template._as_utils_template_with_personalisation(

data.get('personalisation')

)

check_placeholders(template_object)

resp = create_post_template_preview_response(template=template,

template_object=template_object)

return jsonify(resp), 200

开发者ID:alphagov，项目名称:notifications-api，代码行数:27，

示例18: index

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def index():

data = json.loads(request.get_data().decode('utf-8'))

# FETCH THE CRYPTO NAME

crypto_name = data[“nlp”][“entities”][“crypto_name”][0][“raw”]

# FETCH BTC/USD/EUR PRICES

r = requests.get("https://min-api.cryptocompare.com/data/price?fsym="+crypto_name+"&tsyms=BTC,USD,EUR")

return jsonify(

status=200,

replies=[{

'type': 'text',

'content': 'The price of %s is %f BTC and %f USD' % (crypto_name, r.json()['BTC'], r.json()['USD'])

}]

)

开发者ID:TokenBrice，项目名称:sato，代码行数:18，

示例19: index

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def index():

print(port)

data = json.loads(request.get_data().decode('utf-8'))

# FETCH THE CRYPTO NAME

crypto_name = data["nlp"]["entities"]["crypto_name"][0]["raw"]

crypto_ticker = crypto_name.upper()

# FETCH BTC/USD/EUR PRICES

r = requests.get("https://min-api.cryptocompare.com/data/price?fsym="+crypto_ticker+"&tsyms=BTC,USD,EUR")

return jsonify(

status=200,

replies=[{

'type': 'text',

'content': 'The price of %s is :\n%f BTC, \n%f USD, and \n%f EUR.' % (crypto_ticker, r.json()['BTC'], r.json()['USD'], r.json()['EUR'])

}]

)

开发者ID:TokenBrice，项目名称:sato，代码行数:20，

示例20: get_request_validity

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def get_request_validity():

# GitHub signature will suffice for CSRF check

github_signature = request.headers.get("X-Hub-Signature")

if github_signature:

payload_bytes = request.get_data()

for github_webhook_secret in config.github_webhook_secrets:

digest = hmac.new(github_webhook_secret, payload_bytes, sha1).hexdigest()

expected_signature = "sha1=%s" % digest

if expected_signature == github_signature:

return True

# Normal CSRF form tokens work too

token = request.form.get("_csrf_token")

expected_token = session.get("_csrf_token", None)

if expected_token and expected_token == token:

return True

return False

开发者ID:octobear2，项目名称:ob2，代码行数:18，

示例21: set_port_mode

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def set_port_mode(self, switch, interface_id):

"""

Sets the port mode of an interface

:arg str hostname: Hostname or IP of the switch

:arg str interface_id: Interface name (ex. ``FastEthernet0/1``, ``ethernet1/11``)

:body:

``trunk`` or ``access``

"""

mode = request.get_data().lower()

if mode == 'trunk':

switch.set_trunk_mode(interface_id)

elif mode == 'access':

switch.set_access_mode(interface_id)

else:

raise BadRequest('Unknown port mode detected {}'.format(mode))

return 204, None

开发者ID:internap，项目名称:netman，代码行数:21，

示例22: set_bond_port_mode

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def set_bond_port_mode(self, switch, bond_number):

"""

Sets the port mode of a bond

:arg str hostname: Hostname or IP of the switch

:arg int bond_number: Bond number

:body:

``trunk`` or ``access``

"""

mode = request.get_data().lower()

if mode == 'trunk':

switch.set_bond_trunk_mode(bond_number)

elif mode == 'access':

switch.set_bond_access_mode(bond_number)

else:

raise BadRequest('Unknown port mode detected {}'.format(mode))

return 204, None

开发者ID:internap，项目名称:netman，代码行数:21，

示例23: feedback_view_func

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def feedback_view_func(bento_service):

"""

User send feedback along with the request Id. It will be stored and

ready for further process.

"""

if request.content_type != "application/json":

return Response(

response="Incorrect content format, require JSON", status=400

)

data = json.loads(request.get_data().decode("utf-8"))

if "request_id" not in data.keys():

return Response(response="Missing request id", status=400)

if len(data.keys()) <= 1:

return Response(response="Missing feedback data", status=400)

data["service_name"] = bento_service.name

data["service_version"] = bento_service.version

feedback_logger.info(data)

return Response(response="success", status=200)

开发者ID:bentoml，项目名称:BentoML，代码行数:25，

示例24: hit

​点赞 6

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def hit(link):

if '.' in link:

link, ext = link.split('.', 1)

else:

ext = ''

req = '%s %s HTTP/1.1\r\n' % (request.method, request.url)

for k, v in request.headers:

req += '%s: %s\r\n' % (k, v)

req += '\r\n'

req += request.get_data()

createLocalSession()

try:

ret = handlers.target.hit(link, ext, req)

except:

closeLocalSession(True)

raise

else:

closeLocalSession(False)

return ret

开发者ID:daeken，项目名称:SSRFTest，代码行数:23，

示例25: prepare_client_request_data

​点赞 5

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def prepare_client_request_data():

"""

解析出浏览者发送过来的data, 如果是文本, 则进行重写

如果是文本, 则对文本内容进行重写后返回str

如果是二进制则, 则原样返回, 不进行任何处理 (bytes)

:rtype: Union[str, bytes, None]

"""

data = request.get_data() # type: bytes

# 尝试解析浏览器传入的东西的编码

encoding = encoding_detect(data)

if encoding is not None:

try:

data = data.decode(encoding=encoding) # type: str

except:

# 解码失败, data是二进制内容或无法理解的编码, 原样返回, 不进行重写

encoding = None

pass

else:

# data是文本内容, 则进行重写, 并返回str

data = client_requests_text_rewrite(data) # type: str

# 下面这个if是debug用代码, 对正常运行无任何作用

if developer_string_trace: # coverage: exclude

if isinstance(data, str):

data = data.encode(encoding=encoding)

if developer_string_trace.encode(encoding=encoding) in data:

infoprint('StringTrace: appears after client_requests_bin_rewrite, code line no. ', current_line_number())

return data, encoding

开发者ID:aploium，项目名称:zmirror，代码行数:33，

示例26: predict_url_handler

​点赞 5

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def predict_url_handler():

try:

image_url = json.loads(request.get_data())['Url']

results = predict_url(image_url)

return json.dumps(results)

except Exception as e:

print('EXCEPTION:', str(e))

return 'Error processing image'

开发者ID:jamesbannan，项目名称:pluralsight，代码行数:10，

示例27: predict

​点赞 5

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def predict():

# Sending without header

raw = request.get_data()

req = json.loads(raw)

print(req)

result = tf_session.infer(req["path"])

# return "OK"

return jsonify(result)

开发者ID:GoogleCloudPlatform，项目名称:cloudml-edge-automation，代码行数:10，

示例28: predict_url_handler

​点赞 5

​

# 需要导入模块: from flask import request [as 别名]

# 或者: from flask.request import get_data [as 别名]

def predict_url_handler(project=None, publishedName=None):

try:

image_url = json.loads(request.get_data().decode('utf-8'))['url']

results = predict_url(image_url)

return jsonify(results)

except Exception as e:

print('EXCEPTION:', str(e))

return 'Error processing image'

开发者ID:Azure-Samples，项目名称:Custom-vision-service-iot-edge-raspberry-pi，代码行数:10，

注：本文中的flask.request.get_data方法示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。