Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import json
import falcon
import marshmallow as ma
from webargs import fields, ValidationError
from webargs.falconparser import parser, use_args, use_kwargs
from webargs.core import MARSHMALLOW_VERSION_INFO
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)
class Echo(object):
def on_get(self, req, resp):
parsed = parser.parse(hello_args, req)
resp.body = json.dumps(parsed)
adopted_since = IntegerProperty()
class Pet(StructuredNode, models.Node):
"""Pet model"""
pet_id = UniqueIdProperty()
name = StringProperty()
owner = RelationshipFrom("User", "HAS_PET")
class User(StructuredNode, models.Node):
"""User model"""
__validation_rules__ = {
"first_name": fields.Str(),
"last_name": fields.Str(),
"phone_number": fields.Str(required=True)
}
__filtered_fields__ = ["secret_field"]
uid = UniqueIdProperty()
first_name = StringProperty()
last_name = StringProperty()
phone_number = StringProperty(unique_index=True, required=True)
secret_field = StringProperty(default="secret", required=False)
pets = RelationshipTo(Pet, "HAS_PET", model=PetInfo)
class UsersView(GRest):
"""User's View (/users)"""
from webargs.tornadoparser import use_args
from . import messages
from . import util
from .download_file_and_redirect import download_file_and_redirect
from .git_progress import Progress
from .pull_from_remote import pull_from_remote
from .config import Config
thread_pool = ThreadPoolExecutor(max_workers=4)
url_args = {
'file_url': fields.Str(),
'domain': fields.Str(),
'account': fields.Str(),
'repo': fields.Str(),
'branch': fields.Str(),
'path': fields.List(fields.Str()),
'notebook_path': fields.Str(),
}
class LandingHandler(IPythonHandler):
"""
Landing page containing option to download.
Option 1
--------
?file_url=public_file_url
Example: ?file_url=http://localhost:8000/README.md
@use_args({"parent_id": wfields.Str(required=True),
"query": wfields.Str(),
"query_document_id": wfields.Int(),
'metric': wfields.Str(missing='cosine'),
'min_score': wfields.Number(missing=-1),
'max_results': wfields.Int(),
'sort_by': wfields.Str(missing='score'),
'sort_order': wfields.Str(missing='descending',
validate=_is_in_range(['descending',
'ascending'])),
'batch_id': wfields.Int(missing=0),
'batch_size': wfields.Int(missing=10000),
'subset_document_id': wfields.List(wfields.Int()),
})
@marshal_with(SearchResponseSchema())
def post(self, **args):
parent_id = args['parent_id']
def filter_value_type_to_request_arg_type(name, value_type, allow_multiple, load_from=None):
if value_type == str:
arg_type = fields.Str(load_from=load_from or name, location='query')
elif value_type == float:
arg_type = fields.Float(load_from=load_from or name, location='query')
elif value_type == int:
arg_type = fields.Int(load_from=load_from or name, location='query')
elif value_type == datetime:
arg_type = fields.DateTime(load_from=load_from or name, location='query')
else:
raise Exception("Unsupported value type '{}' for a request argument".format(value_type))
if allow_multiple:
arg_type = fields.DelimitedList(arg_type, load_from=load_from or name, location='query')
return arg_type
def get_dbs(cls, query=None, ancestor=None, order=None, limit=None, cursor=None, **kwargs):
args = parser.parse({
'cursor': wf.Str(missing=None),
'limit': wf.Int(missing=None, validate=validate.Range(min=-1)),
'order': wf.Str(missing=None),
})
return util.get_dbs(
query or cls.query(ancestor=ancestor),
limit=limit or args['limit'],
cursor=cursor or args['cursor'],
order=order or args['order'],
**kwargs
)
import neomodel
from flask import Flask
from neomodel import StringProperty, StructuredNode, UniqueIdProperty
from webargs import fields
from grest import GRest, global_config, models, utils
# noinspection PyAbstractClass
class Person(StructuredNode, models.Node):
"""Person model"""
__validation_rules__ = {
"uid": fields.Str(),
"first_name": fields.Str(),
"last_name": fields.Str()
}
uid = UniqueIdProperty()
first_name = StringProperty()
last_name = StringProperty()
class PersonsView(GRest):
"""Person's View (/persons)"""
__model__ = {"primary": Person}
__selection_field__ = {"primary": "uid"}
route_prefix = "/v1" # API version
def create_app():
app = Flask(__name__)
@view_config(..., schema=CanvasOAuthCallbackSchema)
def redirect_uri_view(request):
# The authorization code and state sent by the authorization server
# are available in request.parsed_params.
authorization_code = request.parsed_params["code"]
state = request.parsed_params["state"]
...
This will prevent the view from being called if the code or state is
missing, if the state is invalid or expired, or if there isn't a matching
CSRF token in the session, and it will remove the CSRF token from the
session so that it can't be reused.
"""
location = "querystring"
code = fields.Str(required=True)
state = fields.Str(required=True)
def __init__(self, request):
super().__init__(request)
self.context["secret"] = request.registry.settings["oauth2_state_secret"]
def state_param(self):
"""
Generate and return the value for an OAuth 2 state param.
:rtype: str
"""
request = self.context["request"]
secret = request.registry.settings["oauth2_state_secret"]
csrf = secrets.token_hex()