openstack/deb-python-pecan
Code Issues Proposed changes

Adding validation support.

Browse Source
This commit is contained in:
Jonathan LaCour
2010-10-11 18:09:35 -04:00
parent 1bc441876b
commit b7e444628e
6 changed files with 246 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
pecan/decorators.py
View File

@@ -1,18 +1,37 @@
from inspect import getargspec
def expose(template=None, content_type='text/html'):
def _cfg(f):
if not hasattr(f, 'pecan'): f.pecan = {}
return f.pecan
def expose(template = None,
content_type = 'text/html',
schema = None,
json_schema = None,
error_handler = None):
if template == 'json': content_type = 'application/json'
def decorate(f):
# flag the method as exposed
f.exposed = True
# set a "pecan" attribute, where we will store details
if not hasattr(f, 'pecan'): f.pecan = {}
f.pecan['content_type'] = content_type
f.pecan.setdefault('template', []).append(template)
f.pecan.setdefault('content_types', {})[content_type] = template
cfg = _cfg(f)
cfg['content_type'] = content_type
cfg.setdefault('template', []).append(template)
cfg.setdefault('content_types', {})[content_type] = template
# store the arguments for this controller method
f.pecan['argspec'] = getargspec(f)
cfg['argspec'] = getargspec(f)
# store the validator
cfg['error_handler'] = error_handler
if schema is not None:
cfg['schema'] = schema
cfg['validate_json'] = False
elif json_schema is not None:
cfg['schema'] = json_schema
cfg['validate_json'] = True
return f
return decorate

38
pecan/pecan.py
View File

@@ -4,6 +4,12 @@ from routing import lookup_controller
from webob import Request, Response, exc
from threading import local
from itertools import chain
from formencode import Invalid
try:
from json import loads
except ImportError:
from simplejson import loads
state = local()
@@ -74,13 +80,19 @@ class Pecan(object):
for hook in state.hooks:
getattr(hook, hook_type)(*args)
def get_validated_params(self, all_params, argspec):
def get_params(self, all_params, argspec):
valid_params = dict()
for param_name, param_value in all_params.iteritems():
if param_name in argspec.args:
valid_params[param_name] = param_value
return valid_params
def validate(self, schema, params=None, json=False):
to_validate = params
if json:
to_validate = loads(request.body)
return schema.to_python(to_validate)
def handle_request(self):
# lookup the controller, respecting content-type as requested
# by the file extension on the URI
@@ -104,11 +116,26 @@ class Pecan(object):
# handle "before" hooks
self.handle_hooks('before', state)
# get the result from the controller, properly handling wrap hooks
params = self.get_validated_params(
# fetch and validate any parameters
params = self.get_params(
dict(state.request.str_params),
controller.pecan['argspec']
)
if 'schema' in controller.pecan:
try:
params = self.validate(
controller.pecan['schema'],
json = controller.pecan['validate_json'],
params = params
)
except Invalid, e:
request.validation_error = e
if controller.pecan['error_handler'] is not None:
state.validation_error = e
redirect(controller.pecan['error_handler'])
if controller.pecan['validate_json']: params = dict(data=params)
# get the result from the controller
result = controller(**params)
# pull the template out based upon content type and handle overrides
@@ -142,6 +169,11 @@ class Pecan(object):
state.response = Response()
state.hooks = []
# handle validation errors from redirects
if hasattr(state, 'validation_error'):
state.request.validation_error = state.validation_error
del state.validation_error
# handle the request
try:
self.handle_request()

3
setup.py
View File

@@ -29,7 +29,8 @@ requirements = [
"py >= 1.3.4",
"WebTest >= 1.2.2",
"Paste >= 1.7.5.1",
"PasteScript >= 1.7.3"
"PasteScript >= 1.7.3",
"formencode >= 1.2.2"
]
try:

0
tests/static/test.txt → tests/static/text.txt
View File

4
tests/test_static.py
View File

@@ -16,6 +16,6 @@ class TestStatic(object):
assert response.body == 'Hello, World!'
# get a static resource
response = app.get('/test.txt')
response = app.get('/text.txt')
assert response.status_int == 200
assert response.body == open('tests/static/test.txt', 'rb').read()
assert response.body == open('tests/static/text.txt', 'rb').read()

182
tests/test_validation.py Normal file
View File

@@ -0,0 +1,182 @@
from pecan import Pecan, expose, request, response
from webtest import TestApp
from formencode import validators, Schema
try:
from json import dumps
except ImportError:
from simplejson import dumps
class TestValidation(object):
def test_simple_validation(self):
class RegistrationSchema(Schema):
first_name = validators.String(not_empty=True)
last_name = validators.String(not_empty=True)
email = validators.Email()
username = validators.PlainText()
password = validators.String()
password_confirm = validators.String()
age = validators.Int()
chained_validators = [
validators.FieldsMatch('password', 'password_confirm')
]
class RootController(object):
@expose(schema=RegistrationSchema())
def index(self, first_name,
last_name,
email,
username,
password,
password_confirm,
age):
assert age == 31
assert isinstance(age, int)
return 'Success!'
@expose(json_schema=RegistrationSchema())
def json(self, data):
assert data['age'] == 31
assert isinstance(data['age'], int)
return 'Success!'
# test form submissions
app = TestApp(Pecan(RootController()))
r = app.post('/', dict(
first_name='Jonathan',
last_name='LaCour',
email='jonathan@cleverdevil.org',
username='jlacour',
password='123456',
password_confirm='123456',
age='31'
))
assert r.status_int == 200
assert r.body == 'Success!'
# test JSON submissions
r = app.post('/json', dumps(dict(
first_name='Jonathan',
last_name='LaCour',
email='jonathan@cleverdevil.org',
username='jlacour',
password='123456',
password_confirm='123456',
age='31'
)), [('content-type', 'application/json')])
assert r.status_int == 200
assert r.body == 'Success!'
def test_simple_failure(self):
class RegistrationSchema(Schema):
first_name = validators.String(not_empty=True)
last_name = validators.String(not_empty=True)
email = validators.Email()
username = validators.PlainText()
password = validators.String()
password_confirm = validators.String()
age = validators.Int()
chained_validators = [
validators.FieldsMatch('password', 'password_confirm')
]
class RootController(object):
@expose(schema=RegistrationSchema())
def index(self, first_name,
last_name,
email,
username,
password,
password_confirm,
age):
assert request.validation_error is not None
return 'Success!'
@expose(schema=RegistrationSchema(), error_handler='/errors')
def with_handler(self, first_name,
last_name,
email,
username,
password,
password_confirm,
age):
assert request.validation_error is not None
return 'Success!'
@expose(json_schema=RegistrationSchema())
def json(self, data):
assert request.validation_error is not None
return 'Success!'
@expose(json_schema=RegistrationSchema(), error_handler='/errors')
def json_with_handler(self, data):
assert request.validation_error is not None
return 'Success!'
@expose()
def errors(self, *args, **kwargs):
assert request.validation_error is not None
return 'There was an error!'
# test without error handler
app = TestApp(Pecan(RootController()))
r = app.post('/', dict(
first_name='Jonathan',
last_name='LaCour',
email='jonathan@cleverdevil.org',
username='jlacour',
password='123456',
password_confirm='654321',
age='31'
))
assert r.status_int == 200
assert r.body == 'Success!'
# test with error handler
app = TestApp(Pecan(RootController()))
r = app.post('/with_handler', dict(
first_name='Jonathan',
last_name='LaCour',
email='jonathan@cleverdevil.org',
username='jlacour',
password='123456',
password_confirm='654321',
age='31'
))
assert r.status_int == 302
r = r.follow()
assert r.status_int == 200
assert r.body == 'There was an error!'
# test JSON without error handler
r = app.post('/json', dumps(dict(
first_name='Jonathan',
last_name='LaCour',
email='jonathan@cleverdevil.org',
username='jlacour',
password='123456',
password_confirm='654321',
age='31'
)), [('content-type', 'application/json')])
assert r.status_int == 200
assert r.body == 'Success!'
# test JSON with error handler
r = app.post('/json_with_handler', dumps(dict(
first_name='Jonathan',
last_name='LaCour',
email='jonathan@cleverdevil.org',
username='jlacour',
password='123456',
password_confirm='654321',
age='31'
)), [('content-type', 'application/json')])
assert r.status_int == 302
r = r.follow()
assert r.status_int == 200
assert r.body == 'There was an error!'