Request validation for aiohttp
Project description
aiohttp-d42-validator
Request validation for aiohttp (via d42)
Installation
$ pip3 install aiohttp-d42-validator
Usage
from aiohttp.web import Application, json_response, route, run_app
from d42 import schema
from aiohttp_d42_validator import validate
ParamsSchema = schema.dict({
"q": schema.str.len(1, ...)
})
@validate(params=ParamsSchema)
async def handler(request):
q = request.query["q"]
return json_response({"q": q})
app = Application()
app.add_routes([route("GET", "/users", handler)])
run_app(app)
// http /users?q=Bob
{
"q": "Bob"
}
// http /users
{
"errors": [
"Value <class 'str'> at _['q'] must have at least 1 element, but it has 0 elements"
]
}
Docs
Query params validation
from d42 import schema
from aiohttp_d42_validator import validate
# schema.dict is strict by default (all keys must be present)
ParamsSchema = schema.dict({
"q": schema.str.len(1, ...)
})
@routes.get("/users")
@validate(params=ParamsSchema)
async def handler(request):
q = request.query["q"]
return json_response({"q": q})
Headers validation
from d42 import schema
from aiohttp_d42_validator import validate
from multidict import istr
# "..." means that there can be any other keys
# headers are case-insensitive, so we use istr for
HeadersSchema = schema.dict({
istr("User-Agent"): schema.str.len(1, ...),
...: ...
})
@routes.get("/users")
@validate(headers=HeadersSchema)
async def handler(request):
user_agent = request.headers["User-Agent"]
return json_response({"user_agent": user_agent})
JSON body validation
from d42 import schema
from aiohttp_d42_validator import validate
BodySchema = schema.dict({
"id": schema.int.min(1),
"name": schema.str.len(1, ...),
})
@routes.post("/users")
@validate(json=BodySchema)
async def handler(request):
payload = await request.json()
return json_response({
"id": payload["id"],
"name": payload["name"]
})
URL segments validation
Segments — is a variable part of URL path (aiohttp uses match_info
for it)
from d42 import schema
from aiohttp_d42_validator import validate
SegmentsSchema = schema.dict({
"user_id": schema.str.regex(r"[1-9][0-9]*"),
})
@routes.get("/users/{user_id}")
@validate(segments=SegmentsSchema)
async def handler(request):
user_id = int(request.match_info["user_id"])
return json_response({"user_id": user_id})
Custom response
from http import HTTPStatus
from aiohttp.web import Request, Response
from aiohttp_d42_validator import validate as validate_orig
class validate(validate_orig):
def create_error_response(self, request: Request, errors: List[str]) -> Response:
status = HTTPStatus.UNPROCESSABLE_ENTITY
body = "<ul>" + "".join(f"<li>{error}</li>" for error in errors) + "</ul>"
return Response(status=status, text=body, headers={"Content-Type": "text/html"})
—
Fore more information read d42
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for aiohttp_d42_validator-0.2.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 976a9343d7dee874e882e3d85794849df851ce5bbb3ebbdec890e39f5221ea8b |
|
MD5 | a9ee6370d7b3ef98467ee4ffce5644d8 |
|
BLAKE2b-256 | fe2c5b31035479cd11c905862eb30227080cca95d0e2fffc87adfa8b28d1ffe7 |
Close
Hashes for aiohttp_d42_validator-0.2.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 8084afe24d717d0ad79fe9e9439023250901f29bcd8b60444d1caaaa130411c8 |
|
MD5 | b8a11f5035b2da1f67264b9dab2c7350 |
|
BLAKE2b-256 | 0382579cf3cd922dc866f8f8be50f134b259f8b481c513fa17bad31997be97e2 |