Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def error400(request):
def always_fail(value):
raise ValidationError("something went wrong", status_code=400)
argmap = {"text": fields.Str(validate=always_fail)}
return parser.parse(argmap, request)
def test_nested_many(web_request, parser):
web_request.json = {"pets": [{"name": "Pips"}, {"name": "Zula"}]}
args = {"pets": fields.Nested({"name": fields.Str()}, required=True, many=True)}
parsed = parser.parse(args, web_request)
assert parsed == {"pets": [{"name": "Pips"}, {"name": "Zula"}]}
web_request.json = {}
with pytest.raises(ValidationError):
parser.parse(args, web_request)
def always_error():
def always_fail(value):
raise ma.ValidationError("something went wrong")
args = {"text": fields.Str(validate=always_fail)}
return parser.parse(args)
import json
from flask import Flask, jsonify as J, Response, request
from flask.views import MethodView
import marshmallow as ma
from webargs import fields, ValidationError, missing
from webargs.flaskparser import parser, use_args, use_kwargs
from webargs.core import MARSHMALLOW_VERSION_INFO
class TestAppConfig:
TESTING = True
hello_args = {"name": fields.Str(missing="World", validate=lambda n: len(n) >= 3)}
hello_multiple = {"name": fields.List(fields.Str())}
class HelloSchema(ma.Schema):
name = fields.Str(missing="World", validate=lambda n: len(n) >= 3)
strict_kwargs = {"strict": True} if MARSHMALLOW_VERSION_INFO[0] < 3 else {}
hello_many_schema = HelloSchema(many=True, **strict_kwargs)
app = Flask(__name__)
app.config.from_object(TestAppConfig)
@app.route("/echo", methods=["GET", "POST"])
def echo():
def ModelFilterSet(self, session, models):
class ModelFilterSet(FilterSet):
class Meta:
model = models.Album
query = session.query(models.Album)
operators = (operators.Equal, operators.In)
parser = parser
name__like = Filter('name', fields.Str(), operator=operators.Like)
return ModelFilterSet
from ._utils import route, track, display
blueprint = Blueprint('image', __name__)
cache_filtered = Cache()
cache_unfiltered = Cache(filtered=False)
PLACEHOLDER = "https://raw.githubusercontent.com/jacebrowning/memegen/master/memegen/static/images/missing.png"
OPTIONS = {
'alt': fields.Str(missing=None),
'font': fields.Str(missing=None),
'preview': fields.Bool(missing=False),
'share': fields.Bool(missing=False),
'width': fields.Int(missing=None),
'height': fields.Int(missing=None),
'watermark': fields.Str(missing=None),
}
@blueprint.route("/latest.jpg")
@blueprint.route("/latest.jpg")
@flaskparser.use_kwargs({'filtered': fields.Bool(missing=True)})
def get_latest(index=1, filtered=True):
cache = cache_filtered if filtered else cache_unfiltered
kwargs = cache.get(index - 1)
if kwargs:
kwargs['preview'] = True
else:
kwargs['key'] = 'custom'
kwargs['path'] = "your_meme/goes_here"
kwargs['alt'] = PLACEHOLDER
from flask import Blueprint, current_app, request, redirect
from flask_api import exceptions
from webargs import fields
from ..domain import Text
from ..extensions import cache
from ._parser import parser
from ._utils import route
blueprint = Blueprint('templates', __name__, url_prefix="/api/templates/")
OPTIONS = {
'top': fields.Str(missing=""),
'bottom': fields.Str(missing=""),
'_redirect': fields.Bool(load_from='redirect', missing=True),
'_masked': fields.Bool(load_from='masked', missing=False),
}
@blueprint.route("")
@cache.cached()
def get():
"""Get a list of all meme templates."""
data = OrderedDict()
for template in sorted(current_app.template_service.all()):
url = route('.create', key=template.key, _external=True)
data[template.name] = url
return data
>> serializers = [PeopleSerializer]
>> filters = [ByNameFilter]
>> ...
>> curl -GET /api/people?filter[by-name]=foo
"""
model = None
serializer = None
filters = []
SORT_ARG = 'sort'
FILTER_ARG = 'filter[{}]'
REQUEST_ARGS = {
'sort': fields.DelimitedList(fields.Str(), load_from=SORT_ARG, location='query')
}
def __init__(self, request_helper, *args, **kwargs):
self._request_helper = request_helper
self._api_prefix = kwargs.get('prefix')
self._db = kwargs.get('database')
self._pagination_class = kwargs.get('pagination')
self._response_factory = kwargs.get('response_factory')
self.REQUEST_ARGS.update(self._pagination_class.REQUEST_ARGS)
for filter_cls in self.filters:
filter_user_name = self.FILTER_ARG.format(filter_cls.name.replace('_', '-'))
self.REQUEST_ARGS[filter_cls.name] = self.filter_value_type_to_request_arg_type(filter_user_name,
filter_cls.value_type,
filter_cls.allow_multiple)
from webargs import fields
from ..api.validators import Email, password
user_args = {
'email': fields.Str(validate=Email, required=True),
'password': fields.Str(validate=password, required=True)
}
role_args = {
'name': fields.Str(required=True),
'description': fields.Str(required=True)
}
'reason': fields.Str(required=True),
'allowed_principals': fields.List(fields.Dict(), missing=[])
})
def _process(self, args):
update_blocking(self.blocking, **args)
return jsonify(blocking=blockings_schema.dump(self.blocking, many=False).data)