# aiohttp-d42-validator
[](https://codecov.io/gh/tsv1/aiohttp-d42-validator)
[](https://pypi.python.org/pypi/aiohttp-d42-validator/)
[](https://pypi.python.org/pypi/aiohttp-d42-validator/)
[](https://pypi.python.org/pypi/aiohttp-d42-validator/)
Request validation for [aiohttp](https://docs.aiohttp.org/en/stable/) (via [d42](https://d42.sh))
## Installation
```shell
$ pip3 install aiohttp-d42-validator
```
## Usage
```python
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)
```
```javascript
// http /users?q=Bob
{
"q": "Bob"
}
```
```javascript
// http /users
{
"errors": [
"Value <class 'str'> at _['q'] must have at least 1 element, but it has 0 elements"
]
}
```
## Docs
### Query params validation
```python
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
```python
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
```python
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](https://docs.aiohttp.org/en/stable/web_quickstart.html#variable-resources) `match_info` for it)
```python
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
```python
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](https://d42.sh)
Raw data
{
"_id": null,
"home_page": "https://github.com/tsv1/aiohttp-d42-validator",
"name": "aiohttp-d42-validator",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Nikita Tsvetkov",
"author_email": "tsv1@fastmail.com",
"download_url": "https://files.pythonhosted.org/packages/fe/2c/5b31035479cd11c905862eb30227080cca95d0e2fffc87adfa8b28d1ffe7/aiohttp_d42_validator-0.2.0.tar.gz",
"platform": null,
"description": "# aiohttp-d42-validator\n\n[](https://codecov.io/gh/tsv1/aiohttp-d42-validator)\n[](https://pypi.python.org/pypi/aiohttp-d42-validator/)\n[](https://pypi.python.org/pypi/aiohttp-d42-validator/)\n[](https://pypi.python.org/pypi/aiohttp-d42-validator/)\n\nRequest validation for [aiohttp](https://docs.aiohttp.org/en/stable/) (via [d42](https://d42.sh))\n\n## Installation\n\n```shell\n$ pip3 install aiohttp-d42-validator\n```\n\n## Usage\n\n```python\nfrom aiohttp.web import Application, json_response, route, run_app\nfrom d42 import schema\nfrom aiohttp_d42_validator import validate\n\nParamsSchema = schema.dict({\n \"q\": schema.str.len(1, ...)\n})\n\n@validate(params=ParamsSchema)\nasync def handler(request):\n q = request.query[\"q\"]\n return json_response({\"q\": q})\n\n\napp = Application()\napp.add_routes([route(\"GET\", \"/users\", handler)])\nrun_app(app)\n```\n\n```javascript\n// http /users?q=Bob\n{\n \"q\": \"Bob\"\n}\n```\n\n```javascript\n// http /users\n{\n \"errors\": [\n \"Value <class 'str'> at _['q'] must have at least 1 element, but it has 0 elements\"\n ]\n}\n```\n\n## Docs\n\n### Query params validation\n\n```python\nfrom d42 import schema\nfrom aiohttp_d42_validator import validate\n\n# schema.dict is strict by default (all keys must be present)\nParamsSchema = schema.dict({\n \"q\": schema.str.len(1, ...)\n})\n\n@routes.get(\"/users\")\n@validate(params=ParamsSchema)\nasync def handler(request):\n q = request.query[\"q\"]\n return json_response({\"q\": q})\n```\n\n### Headers validation\n\n```python\nfrom d42 import schema\nfrom aiohttp_d42_validator import validate\nfrom multidict import istr\n\n# \"...\" means that there can be any other keys\n# headers are case-insensitive, so we use istr for\nHeadersSchema = schema.dict({\n istr(\"User-Agent\"): schema.str.len(1, ...),\n ...: ...\n})\n\n@routes.get(\"/users\")\n@validate(headers=HeadersSchema)\nasync def handler(request):\n user_agent = request.headers[\"User-Agent\"]\n return json_response({\"user_agent\": user_agent})\n```\n\n### JSON body validation\n\n```python\nfrom d42 import schema\nfrom aiohttp_d42_validator import validate\n\nBodySchema = schema.dict({\n \"id\": schema.int.min(1),\n \"name\": schema.str.len(1, ...),\n})\n\n@routes.post(\"/users\")\n@validate(json=BodySchema)\nasync def handler(request):\n payload = await request.json()\n return json_response({\n \"id\": payload[\"id\"],\n \"name\": payload[\"name\"]\n })\n```\n\n### URL segments validation\n\nSegments \u2014 is a variable part of URL path (aiohttp [uses](https://docs.aiohttp.org/en/stable/web_quickstart.html#variable-resources) `match_info` for it)\n\n```python\nfrom d42 import schema\nfrom aiohttp_d42_validator import validate\n\nSegmentsSchema = schema.dict({\n \"user_id\": schema.str.regex(r\"[1-9][0-9]*\"),\n})\n\n@routes.get(\"/users/{user_id}\")\n@validate(segments=SegmentsSchema)\nasync def handler(request):\n user_id = int(request.match_info[\"user_id\"])\n return json_response({\"user_id\": user_id})\n```\n\n### Custom response\n\n```python\nfrom http import HTTPStatus\nfrom aiohttp.web import Request, Response\nfrom aiohttp_d42_validator import validate as validate_orig\n\nclass validate(validate_orig):\n def create_error_response(self, request: Request, errors: List[str]) -> Response:\n status = HTTPStatus.UNPROCESSABLE_ENTITY\n body = \"<ul>\" + \"\".join(f\"<li>{error}</li>\" for error in errors) + \"</ul>\"\n return Response(status=status, text=body, headers={\"Content-Type\": \"text/html\"})\n```\n\n\u2014\n\nFore more information read [d42](https://d42.sh)\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Request validation for aiohttp",
"version": "0.2.0",
"project_urls": {
"Docs": "https://d42.sh/docs/integrations/aiohttp",
"GitHub": "https://github.com/tsv1/aiohttp-d42-validator",
"Homepage": "https://github.com/tsv1/aiohttp-d42-validator"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0382579cf3cd922dc866f8f8be50f134b259f8b481c513fa17bad31997be97e2",
"md5": "b8a11f5035b2da1f67264b9dab2c7350",
"sha256": "8084afe24d717d0ad79fe9e9439023250901f29bcd8b60444d1caaaa130411c8"
},
"downloads": -1,
"filename": "aiohttp_d42_validator-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b8a11f5035b2da1f67264b9dab2c7350",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 8255,
"upload_time": "2024-10-13T14:09:54",
"upload_time_iso_8601": "2024-10-13T14:09:54.097422Z",
"url": "https://files.pythonhosted.org/packages/03/82/579cf3cd922dc866f8f8be50f134b259f8b481c513fa17bad31997be97e2/aiohttp_d42_validator-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "fe2c5b31035479cd11c905862eb30227080cca95d0e2fffc87adfa8b28d1ffe7",
"md5": "a9ee6370d7b3ef98467ee4ffce5644d8",
"sha256": "976a9343d7dee874e882e3d85794849df851ce5bbb3ebbdec890e39f5221ea8b"
},
"downloads": -1,
"filename": "aiohttp_d42_validator-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "a9ee6370d7b3ef98467ee4ffce5644d8",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 8667,
"upload_time": "2024-10-13T14:09:55",
"upload_time_iso_8601": "2024-10-13T14:09:55.807952Z",
"url": "https://files.pythonhosted.org/packages/fe/2c/5b31035479cd11c905862eb30227080cca95d0e2fffc87adfa8b28d1ffe7/aiohttp_d42_validator-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-10-13 14:09:55",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "tsv1",
"github_project": "aiohttp-d42-validator",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [
{
"name": "aiohttp",
"specs": [
[
">=",
"3.5"
],
[
"<",
"4.0"
]
]
},
{
"name": "multidict",
"specs": [
[
"<",
"7.0"
],
[
">=",
"4.5"
]
]
},
{
"name": "d42",
"specs": [
[
"<",
"3.0"
],
[
">=",
"2.0"
]
]
}
],
"lcname": "aiohttp-d42-validator"
}
JSON
