Web use¶
Introduction¶
Let’s see a simple way to use trafaret
for handle endpoints of rest api in
aiohttp
.
We’ll create a simple rest api for create / update books. The first thing which
we need to do it’s describe our schemes of input data in trafaret format.
When we work with form in the web we receive all data in a string format. So when you want to send a boolean type or list of integers you send some like this.
"True" # True
"1, 2, 3" # [1, 2, 3]
Trafaret designed for solve these problems.
import trafaret as t
def comma_to_list(text):
"""Convert string with words separated by comma to list."""
return [
s.strip() for s in text.split(',')
]
create_book_chacker = t.Dict({
'title': t.String,
'authors': comma_to_list,
'sold': t.StrBool,
})
So, when you recive data from form it’s not problem for you, because StrBool
and comma_to_list
prepare data for you in correct format.
create_book_chacker.check({"title": 'Glue', 'authors': 'Welsh,', 'sold': 'True'})
# {'title': 'Glue', 'authors': ['Welsh', ''], 'sold': True}
But if you receive data as json
it’s not very useful for you, because you
can receive data in correct format from the client.
The second problem which trafaret solved it’s a camel/snake case war. People why write in python prefe use snake_case unlike people why write in ES and use CamelCase. Trafaret give an approach for rename key of dictionary for solve this problem.
t.Dict({t.Key('userNameFirst') >> 'first_name': t.String})
# or
t.Dict({t.Key('userNameFirst', to_name='first_name'): t.String})
Schemes¶
So, now we are ready to write our schemes with trafaret. We can put this to
the utils.py
.
import trafaret as t
create_book_chacker = t.Dict({
t.Key('bookTitle', to_name='title'): t.String,
t.Key('bookPageCount', to_name='pages'): t.Int,
t.Key('bookDescription', to_name='description'): t.String(min_length=20),
t.Key('bookPrice', to_name='price', default=100): t.Int >= 100,
t.Key('bookIsFree', optional=True, to_name='is_free'): t.Bool,
t.Key('bookFirstAuthor', to_name='first_author'): t.String(max_length=10),
t.Key('bookAuthors', to_name='authors'): t.List(t.String(max_length=10)),
})
update_user_chacker = create_book_chacker + {"id": t.Int}
Here we created a two schemes. For validate data which need to create a book
and for update. This two schemes differing only by id
field.
After that we can use this checkers for validation data in our web handlers. But for allocation all logic which connected with trafaret let’s create functions which do it.
def prepare_data_for_create_book(data):
valid_data = create_book_chacker.check(data)
# do something else
...
return valid_data
def prepare_data_for_update_book(data):
valid_data = update_user_chacker.check(data)
# do something else
...
return valid_data
Handlers¶
Let’s use these function in our handlers.
from aiohttp import web
# handlers
async def create_book(req):
"""Hadler for create book"""
raw_data = await req.json()
data = prepare_data_for_create_book(raw_data)
# do something
...
return web.json_response({"created": True})
async def update_book(req):
"""Handler for update book by id"""
raw_data = await req.json()
data = prepare_data_for_update_book(raw_data)
# do something
...
return web.json_response({"updated": True})
# setup an application
app = web.Application()
app.add_routes([
web.post('/', create_book),
web.put('/', update_book)
])
web.run_app(app, port=8000)
After that we can send request to the our server.
import requests as r
data = {
"bookTitle": "Glue",
"bookPageCount": 436,
"bookDescription": "Glue tells the stories of four Scottish boys over four decades...",
"bookPrice": 423,
"bookFirstAuthor": "Welsh",
"bookAuthors": ["Welsh"]
}
r.post("http://0.0.0.0:8000/", json=data).text
# '{"created": true}'
Errors¶
We made validation for input data but also we want eazy show errors if we have problem with it.
If input data is not valid then trafaret after call check method raise error
(t.DataError
) connected with that. Let’s see easy way to handle all errors
connected with trafater
.
from functools import wraps
def with_error(fn):
"""
This is decorator for wrapping web handlers which need to represent
errors connected with validation if they exist.
"""
@wraps(fn)
async def inner(*args, **kwargs):
try:
return await fn(*args, **kwargs)
except t.DataError as e:
return web.json_response({
'errors': e.as_dict(value=True)
})
return inner
After that we need to wrap all our handlers.
@with_error
async def create_book(req):
"""Hadler for create book""""
...
@with_error
async def update_book(req):
"""Handler for update book by id"""
...
That is it. Now, we receive pretty error messages when our input data is not valid.
import requests as r
data = {
"bookTitle": "Glue",
"bookPageCount": 436,
"bookDescription": "Glue tells the stories of four Scottish boys over four decades...",
"bookPrice": 423,
"bookFirstAuthor": "Welsh",
"bookAuthors": ["Welsh"]
}
r.put("http://0.0.0.0:8000/", json=data).text
# '{"errors": {"id": "is required"}}'