Rapidy

What's new:

This update is very extensive indeed.

Functional changes
Optional request attribute
The request attribute is no longer mandatory for functional handlers. If the request needs to be utilized, simply declare an attribute and assign it the type web.Request. `rAPIdy` will then automatically inject the request into the handler via annotation."
python
from typing import Annotated
from rapidy import web

routes = web.RouteTableDef()

routes.get('/default_handler_example')
async def default_handler_example(
request: str = web.Request,
) -> web.Response:
print({'request': request})
return web.json_response({'data': 'success'})

routes.get('/handler_without_request_example')
async def handler_without_request_example() -> web.Response:
return web.json_response({'data': 'success'})

routes.get('/handler_request_as_snd_attr_example')
async def handler_request_as_snd_attr_example(
host: Annotated[str, web.Header(alias='Host')],
request: web.Request,
) -> web.Response:
print({'host': host, 'request': request})
return web.json_response({'data': 'success'})

app = web.Application()
app.add_routes(routes)

if __name__ == '__main__':
web.run_app(app, host='127.0.0.1', port=8080)


Parameter definition in two ways
Parameters can now be defined in two ways:
* By using the `Annotated` type
* By passing the parameter as a default attribute value

> [!WARNING]
> If you're using `mypy`, be sure to enable the plugin.
python
async def handler(
header_param_1: str = web.Header(),
header_param_2: Annotated[str, web.Header()],
) -> web.Response:


Default values
Added support for default values for `rAPIdy-parameters`
python
async def handler(
header_param_1: str = web.Header('default'),
header_param_2: Annotated[str, web.Header()] = 'default',

cookie_param_1: str = web.Cookie('default'),
cookie_param_2: Annotated[str, web.Cookie()] = 'default',

query_param_1: str = web.Query('default'),
query_param_2: Annotated[str, web.Query()] = 'default',

json_param_1: str = web.JsonBody('default'),
json_param_2: Annotated[str, web.JsonBody()] = 'default',
) -> web.Response:


Server information hiding flag
We have introduced a flag to hide server information, as we consider its visibility as a potential security vulnerability

python
app = web.Application(server_info_in_response=True) default = False


Intercepting custom errors
Now you can capture custom errors with `web_exceptions.HTTPValidationFailure.validation_errors`.
python
import logging
from rapidy import web
from rapidy.typedefs import Handler
from rapidy.web_exceptions import HTTPValidationFailure

logger = logging.getLogger(__name__)

routes = web.RouteTableDef()

routes.get('/')
async def handler() -> web.Response:
return web.json_response({'data': 'success'})

web.middleware
async def error_catch_middleware(request: web.Request, handler: Handler) -> web.StreamResponse:
try:
return await handler(request)

except HTTPValidationFailure as validation_failure_error:
client_errors = validation_failure_error.validation_errors
logger.error('Client error catch, errors: %s', client_errors)
raise validation_failure_error

except Exception as unhandled_error:
logger.error('Unhandled error: %s', unhandled_error)
return web.json_response(status=500)

app = web.Application(middlewares=[error_catch_middleware])
app.add_routes(routes)

if __name__ == '__main__':
web.run_app(app, host='127.0.0.1', port=8080)


Improved multipart type
Now multipart can retrieve fields that do not have a Content-Type header, such fields will be retrieved as a bytearray
and passed to pydantic for validation.

Mypy support
Added support for mypy:
* Added plugin for mypy
* Added py.typed file - now mypy will be able to see and correctly define library types
toml
example for pyproject.toml
...
[tool.mypy]
plugins = [
"pydantic.mypy",
"rapidy.mypy"
]
...


Refactoring
Improved parameters
We have made slight modifications to the parameters to enhance ease of initialization and removed fields that related to internal logic. As a result, IDEs will now provide more accurate suggestions about which attributes are modifiable

Added checks for some incorrect user scripts
We have made minor changes to the internal workings of the library and implemented checks for certain erroneous user scripts

General changes
* We have extended and expanded README.md for your convenience
* Contributors file has been added
* A license file has been included

`Our team sincerely thanks our users for choosing our product. We are committed to providing the best experience possible for you`

Feature / Bugfix
* ✅ **middleware** now allows query validation
python
from rapidy.web import middleware

middleware
async def auth_middleware(
request: web.Request,
handler: HandlerType,
bearer_token: Annotated[str, Header(alias='Authorization')],
) -> web.StreamResponse:
assert bearer_token == BEARER_TOKEN
return await handler(request)

middleware
async def request_id_middleware(
request: web.Request,
handler: HandlerType,
request_id: Annotated[str, Header(alias='Request-ID')],
) -> web.StreamResponse:
return await handler(request)

async def protected_handler(request: web.Request, body: Annotated[str, TextBody]) -> web.Response:
return web.Response()

api_app = web.Application(middlewares=[auth_middleware])
api_app.add_routes([web.post('/protected_handler', protected_handler)])

app = web.Application(middlewares=[request_id_middleware])
app.add_subapp('/security', api_app)

rAPIdy 💮
**write quickly** 🚀 **write beautifully** 🌸

<a href="https://github.com/daniil-grois/rAPIdy">rAPIdy<a/> is a minimalistic, asynchronous, fast web framework based in aiohttp and pydantic.


pip install rapidy


Key Features:
* ✏️ easy to write and read code
* 📔 <a href="https://github.com/pydantic/pydantic">pydantic<a/> native support -> **yes** we support **V**1️⃣ and **V**2️⃣❗
* 🚀 <a href="https://github.com/aio-libs/aiohttp">aiohttp<a/> based and native features support - aiohttp one of the fastest and most productive HTTP servers
* 📤 convenient support for basic types of incoming data extraction

Quickstart

Create endpoint using RouteTableDef

rAPIdy inherits the basic functionality of aiohttp <a href="https://docs.aiohttp.org/en/stable/web_quickstart.html">quickstart</a>

python
from rapidy import web
from typing_extensions import Annotated

routes = web.RouteTableDef()


routes.get('/hello')
async def hello(
request: web.Request,
username: Annotated[str, web.JsonBody(min_length=3, max_length=20)],
password: Annotated[str, web.JsonBody(min_length=8, max_length=40)],
) -> web.Response:
write you business code here
return web.json_response({'username': username, 'password': password})


app = web.Application()
app.add_routes(routes)

if __name__ == '__main__':
web.run_app(app, port=8000)


Create endpoint using web.<method_name>

python
from rapidy import web
from typing_extensions import Annotated


async def hello(
request: web.Request,
username: Annotated[str, web.JsonBody(min_length=3, max_length=20)],
password: Annotated[str, web.JsonBody(min_length=8, max_length=40)],
) -> web.Response:
write you business code here
return web.json_response({'username': username, 'password': password})


app = web.Application()
app.add_routes([web.post('/hello', hello)])

if __name__ == '__main__':
web.run_app(app, port=8000)


Create endpoint using web.view

python
from rapidy import web
from typing_extensions import Annotated


class Handler(web.View):
async def post(
self,
username: Annotated[str, web.JsonBody(min_length=3, max_length=20)],
password: Annotated[str, web.JsonBody(min_length=8, max_length=40)],
) -> web.Response:
write you business code here
return web.json_response({'username': username, 'password': password})


app = web.Application()
app.add_routes([web.view('/hello', Handler)])

if __name__ == '__main__':
web.run_app(app, port=8000)


Pydantic native support

python
from rapidy import web
from typing_extensions import Annotated


async def hello_handler(
request: web.Request,
username: Annotated[str, web.JsonBody(min_length=3, max_length=20)],
password: Annotated[str, web.JsonBody(min_length=8, max_length=40)],
) -> web.Response:
write you business code here
return web.json_response({'username': username, 'password': password})


app = web.Application()
app.add_routes([web.post('/hello', hello_handler)])
web.run_app(app, port=8000)

✅✅✅ Success request validation ✅✅✅

curl -X POST \
-H "Content-Type: application/json" -d '{"username": "Max", "password": "myAwesomePass"}' -v \
http://127.0.0.1:8000/hello

< HTTP/1.1 200 OK ... {"username": "Max", "password": "myAwesomePass"}


❌❌❌ Request validation failure ❌❌❌


curl -X POST \
-H "Content-Type: application/json" -d '{"username": "M", "password": "m"}' -v \
http://127.0.0.1:8000/hello

< HTTP/1.1 422 Unprocessable Entity ...
{
"errors": [
{
"loc": ["body", "username"],
"type": "string_too_short",
"msg": "String should have at least 3 characters",
"ctx": {"min_length": 3}
},
{
"type": "string_too_short",
"loc": ["body", "password"],
"msg": "String should have at least 8 characters",
"ctx": {"min_length": 8}
}
]
}



Choose your path

* You can create APIs the way you want.
* **rAPIdy** supports 3 basic types for defining incoming parameters
* 🌒 param
* Path
* Header
* Cookie
* Query
* BodyJson
* FormDataBody
* MultipartBody
* 🌕 schema
* PathSchema
* HeaderSchema
* CookieSchema
* QuerySchema
* BodyJsonSchema
* FormDataBodySchema
* MultipartBodySchema
* 🌑 raw data (no validate with pydantic)
* PathRaw - **Dict[str, str]**
* HeaderRaw - **Dict[str, str]**
* CookieRaw - **Dict[str, str]**
* QueryRaw - **Dict[str, str]**
* BodyJsonRaw - **Dict[str, Any]**
* FormDataBodyRaw - **Dict[str, Any]**
* MultipartBodyRaw - **Dict[str, Any]**
* TextBody - **str**
* BytesBody - **bytes**
* StreamBody - **aiohttp.streams.StreamReader**

python
defining request attributes as param 🌒
from rapidy import web
from typing_extensions import Annotated


async def hello_handler(
request: web.Request,
path params
path_param: Annotated[str, web.Path],
headers
host: Annotated[str, web.Header(alias='Host')],
user_agent: Annotated[str, web.Header(alias='User-Agent')],
cookie
user_cookie1: Annotated[str, web.Cookie(alias='UserCookie1')],
user_cookie2: Annotated[str, web.Cookie(alias='UserCookie2')],
query params
user_param1: Annotated[str, web.Query(alias='UserQueryParam1')],
user_param2: Annotated[str, web.Cookie(alias='UserQueryParam2')],
body
username: Annotated[str, web.JsonBody(min_length=3, max_length=20)],
password: Annotated[str, web.JsonBody(min_length=8, max_length=40)],
) -> web.Response:
write you business code here
...
return web.Response()


app = web.Application()
app.add_routes([web.post('/hello/{path_param}', hello_handler)])

python
defining request attributes as schema 🌕
from pydantic import BaseModel, Field

class PathRequestSchema(BaseModel):
path_param: str

class HeaderRequestSchema(BaseModel):
host: str = Field(alias='Host')
user_agent: str = Field(alias='User-Agent')

class CookieRequestSchema(BaseModel):
user_cookie1: str = Field(alias='UserCookie1')
user_cookie2: str = Field(alias='UserCookie2')

class QueryRequestSchema(BaseModel):
user_cookie1: str = Field(alias='UserQueryParam1')
user_cookie2: str = Field(alias='UserQueryParam1')

class BodyRequestSchema(BaseModel):
username: str = Field(min_length=3, max_length=20)
password: str = Field(min_length=8, max_length=40)

async def hello_handler(
request: web.Request,
path: Annotated[PathRequestSchema, web.PathSchema],
headers: Annotated[HeaderRequestSchema, web.HeaderSchema],
cookies: Annotated[CookieRequestSchema, web.Cookie],
query: Annotated[QueryRequestSchema, web.QuerySchema],
body: Annotated[BodyRequestSchema, web.JsonBodySchema],
) -> web.Response:



python
defining request attributes as raw 🌑
async def hello_handler(
request: web.Request,
path: Annotated[Dict[str, str], web.PathRaw],
headers: Annotated[Dict[str, str], web.HeaderRaw],
cookies: Annotated[Dict[str, str], web.CookieRaw],
query: Annotated[Dict[str, str], web.QueryRaw],
body: Annotated[Dict[str, Any], web.JsonBodyRaw],
) -> web.Response:


python
also you may to combine 🌒 🌕 🌑
async def hello_handler(
request: web.Request,
path_param: Annotated[str, web.Path],
headers: Annotated[Dict[str, str], web.HeaderRaw],
body: Annotated[BodyRequestSchema, web.JsonBodySchema],
) -> web.Response: