diff --git a/tests/func_test.py b/tests/func_test.py index d431d5c..a89c0f3 100644 --- a/tests/func_test.py +++ b/tests/func_test.py @@ -1,5 +1,5 @@ -import sys import json +import sys from jsonschema.validators import Draft4Validator, create from tornado.testing import AsyncHTTPTestCase @@ -62,6 +62,27 @@ def post(self): return self.body['name'] +def get_countries(): + return ['Brazil', 'Argentina'] + + +class MemberHandler(requesthandlers.APIHandler): + """Example handler with input schema validation that uses SchemaCallback + """ + @schema.validate( + input_schema={ + "type": "object", + "properties": { + "name": {'type': "string"}, + "country": {'enum': schema.SchemaCallback(get_countries)} + }, + "required": ['name', 'country'], + } + ) + def post(self): + return self.body['name'] + + class DBTestHandler(requesthandlers.APIHandler): """APIHandler for testing db_conn""" def get(self): @@ -130,11 +151,12 @@ class APIFunctionalTest(AsyncHTTPTestCase): def get_app(self): rts = routes.get_routes(helloworld) rts += [ + ("/api/member", PeopleHandler), ("/api/people", PeopleHandler), ("/api/explodinghandler", ExplodingHandler), ("/api/notfoundhandler", NotFoundHandler), ("/views/someview", DummyView), - ("/api/dbtest", DBTestHandler) + ("/api/dbtest", DBTestHandler), ] return application.Application( routes=rts, @@ -142,11 +164,23 @@ def get_app(self): db_conn=None ) + def test_post_schemacallback(self): + """ Test if SchemaCallback will be evalued + """ + r = self.fetch( + "/api/people", + method="POST", + body=jd({ + 'name': "Paulo", + }) + ) + self.assertEqual(r.code, 400) + def test_post_custom_validator_class(self): """It should not raise errors because ExtendedDraft4Validator is used, so schema type 'int' is allowed. """ r = self.fetch( - "/api/people", + "/api/member", method="POST", body=jd({ 'name': "Paulo", diff --git a/tests/test_schema.py b/tests/test_schema.py new file mode 100644 index 0000000..2d1d3b4 --- /dev/null +++ b/tests/test_schema.py @@ -0,0 +1,95 @@ +import unittest + +from tornado_json.schema import SchemaCallback +from tornado_json.schema import SchemaHelper +from tornado_json.schema import detect_schemacallback +from tornado_json.schema import evaluate_schema_callbacks + + +class MySchema(SchemaHelper): + + def get_enum_key(self, requesthandler): + return ['A', 'B'] + + +class MyNestedSchema(SchemaHelper): + + def get_properties_key(self, requesthandler): + return { + 'myschema': MySchema(), + 'memory_length': { + 'type': "integer", + } + } + + def get_title_key(self, requesthandler): + return "My Nested Schema" + + +class TestSchemaHelper(unittest.TestCase): + + def test_schemahelper(self): + sh = MySchema() + self.assertEqual( + sh.as_dict(), + {'enum': ["A", "B"]}) + + def test_schemahelper_nested(self): + sh = MyNestedSchema() + self.assertEqual( + sh.as_dict(), + { + 'title': "My Nested Schema", + 'properties': { + 'myschema': { + 'enum': ["A", "B"], + }, + 'memory_length': { + 'type': "integer", + }, + }, + }) + + +class TestSchemaCallback(unittest.TestCase): + + def test_evaluate_schema_callbacks(self): + + def get_type(): + return ['object', 'string'] + + scb = SchemaCallback(get_type) + schema = yield evaluate_schema_callbacks({'type': scb}) + self.assertEqual(schema, + {'type': ['object', 'string']}) + + def test_detect_schemacallback(self): + + def get_users(table, role): + pks = [1, 10, 100] + return pks + + scb = SchemaCallback(get_users, "users", role="admin") + + self.assertTrue(detect_schemacallback({ + 'properties': { + 'database': { + 'type': 'object', 'properties': {'database': scb}}}})) + + self.assertFalse(detect_schemacallback({ + 'properties': { + 'database': {'type': 'object', 'properties': {}}}})) + + def test_schemacallback(self): + + def get_users(table, role): + pks = [1, 10, 100] + return pks + + scb = SchemaCallback(get_users, "users", role="admin") + result = yield scb() + self.assertEqual(result, [1, 10, 100]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tornado_json/schema.py b/tornado_json/schema.py index 8887db8..25f8ea7 100644 --- a/tornado_json/schema.py +++ b/tornado_json/schema.py @@ -1,11 +1,16 @@ import json + +try: + from UserDict import UserDict +except ImportError: + from collections import UserDict + from functools import wraps import jsonschema import tornado.gen - from tornado_json.exceptions import APIError try: @@ -18,6 +23,116 @@ from tornado_json.utils import container +def schemahelpers_as_dict(schema, requesthandler): + """Return schema copy with all (nested too) SchemaHelper processed as a + dict.""" + d = {} + for k, v in schema.items(): + if isinstance(v, SchemaHelper): + d[k] = v.as_dict(requesthandler) + elif isinstance(v, dict): + d[k] = schemahelpers_as_dict(v, requesthandler) + else: + d[k] = v + return d + + +class SchemaHelper(UserDict): + """Helper class to create dynamic schemas in an elegant way. + + SchemaHelper is a just an extend dict with method "as_dict", this methods + returns the dict itself updated the result of all methods that match the + pattern above: + + get_%(key)s_key as dict[%(key)s] + + :Example: + + >>> class MySchema(SchemaHelper): + ... def get_type_key(self): + ... return "object" + >>> + >>> myschema = MySchema() + >>> myschema.as_dict() + {"type": "object"} + """ + + def as_dict(self, requesthandler=None): + """Return a dict containing the SchemaHelper processed methods. + + :param requesthandler: Tornado request handler instance + :type tornado.web.RequestHandler: + :returns: dict to be used on jsonschema + :rtype: dict + """ + d = {} + + for attrname in dir(self): + if attrname.startswith("get_") and attrname.endswith("_key"): + method = getattr(self, attrname) + if callable(method): + k = attrname[4:-4] + v = method(requesthandler) + if isinstance(v, SchemaHelper): + v = v.as_dict(requesthandler) + d[k] = v + + d = schemahelpers_as_dict(d, requesthandler) + + return d + + +@tornado.gen.coroutine +def evaluate_schema_callbacks(schema): + """Return schema with all schema callbacks evaluated. + + :param schema: schema with callback(s) + :type schema: dict + :returns: schema evaluated + :rtype: dict + """ + d = {} + for k, v in schema.items(): + if isinstance(v, SchemaCallback): + v = v() + elif isinstance(v, dict): + v = yield evaluate_schema_callbacks(v) + d[k] = v + raise tornado.gen.Return(d) + + +def detect_schemacallback(schema): + """Check recursively if the schema has a SchemaCallback instance.""" + for k, v in schema.items(): + if isinstance(v, SchemaCallback): + return True + elif isinstance(v, dict): + if detect_schemacallback(v): + return True + + return False + + +class SchemaCallback(object): + """Class to create dynamic key values on schemas.""" + + def __init__(self, func, *args, **kwargs): + """Store the callable with args and kwargs. + + :param func: callable method + :param args: args to be passsed to func + :params kwargs: kwargs to be passed to func + :type func: callable + """ + self.func = func + self.args = args + self.kwargs = kwargs + + def __call__(self): + """Call the function asynchronously trought tornado.gen.Task class.""" + return self.func(*self.args, **self.kwargs) + + def validate(input_schema=None, output_schema=None, input_example=None, output_example=None, validator_cls=None, @@ -30,6 +145,11 @@ def validate(input_schema=None, output_schema=None, :param on_empty_404: If this is set, and the result from the decorated method is a falsy value, a 404 will be raised. """ + + input_schema_has_callback = False + if input_schema is not None: + input_schema_has_callback = detect_schemacallback(input_schema) + @container def _validate(rh_method): """Decorator for RequestHandler schema validation @@ -57,6 +177,7 @@ def _wrapper(self, *args, **kwargs): # In case the specified input_schema is ``None``, we # don't json.loads the input, but just set it to ``None`` # instead. + input_schema = _wrapper.input_schema if input_schema is not None: # Attempt to json.loads the input try: @@ -69,6 +190,14 @@ def _wrapper(self, *args, **kwargs): raise jsonschema.ValidationError( "Input is malformed; could not decode JSON object." ) + + # if isinstance(input_schema, SchemaHelper): + # input_schema = input_schema.as_dict(self, *args, **kw) + + if input_schema_has_callback: + input_schema = \ + yield evaluate_schema_callbacks(input_schema) + # Validate the received input jsonschema.validate( input_,