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
File details
Details for the file aiohttp_d42_validator-0.2.0.tar.gz
.
File metadata
- Download URL: aiohttp_d42_validator-0.2.0.tar.gz
- Upload date:
- Size: 8.7 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.1 CPython/3.12.7
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 976a9343d7dee874e882e3d85794849df851ce5bbb3ebbdec890e39f5221ea8b |
|
MD5 | a9ee6370d7b3ef98467ee4ffce5644d8 |
|
BLAKE2b-256 | fe2c5b31035479cd11c905862eb30227080cca95d0e2fffc87adfa8b28d1ffe7 |
File details
Details for the file aiohttp_d42_validator-0.2.0-py3-none-any.whl
.
File metadata
- Download URL: aiohttp_d42_validator-0.2.0-py3-none-any.whl
- Upload date:
- Size: 8.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.1 CPython/3.12.7
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 8084afe24d717d0ad79fe9e9439023250901f29bcd8b60444d1caaaa130411c8 |
|
MD5 | b8a11f5035b2da1f67264b9dab2c7350 |
|
BLAKE2b-256 | 0382579cf3cd922dc866f8f8be50f134b259f8b481c513fa17bad31997be97e2 |