Skip to main content

Request validation for aiohttp

Project description

aiohttp-d42-validator

Codecov PyPI PyPI - Downloads Python Version

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


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aiohttp_d42_validator-0.2.0.tar.gz (8.7 kB view hashes)

Uploaded Source

Built Distribution

aiohttp_d42_validator-0.2.0-py3-none-any.whl (8.3 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page