-
Notifications
You must be signed in to change notification settings - Fork 9
Examples
Danila Ganchar edited this page Oct 5, 2024
·
11 revisions
Headers validation works first. All other parameters will processed only after successful validation of headers. Example:
@app.route('/headers')
@validate_params(
Param('Accept-Language', HEADER, str, rules=[Enum('en-US', 'be-BE')])
)
def headers(valid: ValidRequest):
return jsonify(valid.get_headers())
Input | Output |
---|---|
curl http://localhost:5000/headers |
🔴 invalid header Accept-Language. value is required |
curl -H 'Accept-Language: test' http://localhost:5000/headers |
🔴 invalid header Accept-Language. not allowed, allowed values: en-US |
curl -H 'Accept-Language: be-BE' http://localhost:5000/headers |
🟢 {"Accept-Language": "be-BE"} |
In some cases we need to use complex logic or third-party services for validation. You can create a custom validation rule:
class CountryCodeRule(AbstractRule):
def validate(self, value: str) -> str:
# do here all what you need...
data = requests.get(f'https://restcountries.com/v3.1/alpha/{value}').json()
if isinstance(data, list) and len(data) > 0:
# correct value
return value
raise RuleError(f'invalid country code: {value}')
@app.route('/country/<string:country_code>')
@validate_params(
Param('country_code', PATH, str, rules=[CountryCodeRule()]),
)
def country(valid: ValidRequest, country_code):
return jsonify(country_code=country_code)
Request / Response examples:
curl http://localhost:5000/country/by # {"country_code": "by"}
curl http://localhost:5000/country/unknown_country_code
# {'path': {'country_code': RulesError(RuleError('invalid country code: unknown_country_code'))}}
In some cases we need additional processing only after all types and values are correct. You can create a custom AfterParam
:
class VersionAfterParam(AbstractAfterParam):
def validate(self, value: ValidRequest):
# do here all what you need...
data = value.get_json()
if 'some_flag' in data and int(data['version'][1:]) < 99:
raise AfterParamError('some_flag is only supported starting from version 99')
@app.route('/version', methods=['POST'])
@validate_params(
Param('some_flag', JSON, str, required=False),
Param('version', JSON, str, rules=[Enum('v1', 'v2', 'v99')]),
VersionAfterParam(),
)
def version(valid: ValidRequest):
return jsonify(version=valid.get_json().get('some_flag'))
Request / Response examples:
curl -H 'Content-Type: application/json' -d '{"version": "v1", "some_flag": "test"}' http://localhost:5000/version
# some_flag is only supported starting from version 99
curl -H 'Content-Type: application/json' -d '{"version": "v99", "some_flag": "test"}' http://localhost:5000/version
# {"some_flag": "test"}
from flask import Flask, jsonify
from flask_request_validator import *
from flask_request_validator import JsonParam as P
from flask_request_validator.error_formatter import demo_error_formatter
from flask_request_validator.exceptions import InvalidRequestError
app = Flask(__name__)
@app.errorhandler(InvalidRequestError)
def data_error(e):
return jsonify(demo_error_formatter(e))
@app.route('/json', methods=['POST'])
@validate_params(P(dict(person=P(
dict(info=P(
dict(
# just string with dt
dt=[Datetime('%Y-%m-%d')],
contacts=P(
dict(
# list with strings
phones=P([Enum('+375', '+49')], as_list=True),
# list with dicts
networks=P(
dict(name=[Enum('facebook', 'telegram')]), as_list=True,
),
# list with strings
emails=P([IsEmail()], as_list=True),
),
),
)
)))))
)
def json(valid: ValidRequest):
return jsonify({'json': valid.get_json()})
call:
curl --header "Content-Type: application/json" \
--request POST \
--data '{"person":{"info":{"dt":"invalid dt","contacts":{"phones":["+375","bad", "+49", "bad2"],"networks":[{"name":"insta"}],"emails":["[email protected]","bad"]}}}}' \
http://localhost:5000/json
result:
{'json': [JsonError(['root', 'person', 'info', 'contacts', 'phones'], {1: RulesError(ValueEnumError(('+375', '+49'))), 3: RulesError(ValueEnumError(('+375', '+49')))}, True), JsonError(['root', 'person', 'info', 'contacts', 'networks'], {0: {'name': RulesError(ValueEnumError(('facebook', 'telegram')))}}, True), JsonError(['root', 'person', 'info', 'contacts', 'emails'], {1: RulesError(ValueEmailError())}, True), JsonError(['root', 'person', 'info'], {'dt': RulesError(ValueDatetimeError('%Y-%m-%d'))}, False)]}
You can use JsonParam
without @validate_params
:
json_data = {}
param = JsonParam(...)
valid_data, errors = param.validate(json_data) # type: List[JsonError]
from flask import Flask
from flask_request_validator import *
app = Flask(__name__)
@app.errorhandler(RequestError)
def handler(e):
return str(e.files), 400
@app.route('/issue/85', methods=['POST'])
@validate_params(
File(
mime_types=['application/pdf'],
max_size=5 * 1000 * 1000,
name='document',
),
File(
mime_types=['image/jpeg'],
max_size=5 * 1000 * 1000,
name='photo',
)
)
def issue_85(valid: ValidRequest):
return str(valid.get_flask_request().files.keys())
Request / Response example:
curl http://localhost:5000/issue/85 -F "document=@/FULL_PATH_TO.torrent"
# {'files': [FileMimeTypeError('document', 'application/octet-stream', ['application/pdf']), FileMissingError('photo')]}