Skip to content

Examples

Danila Ganchar edited this page Jan 15, 2024 · 7 revisions

Headers validation

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"}

Custom validation Rule

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'))}}

AfterParam hook

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 
# {"version": "test"}

JsonParam - Nested JSON validation

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":["valid@gmail.com","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]

Files validation

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')]}