R Ap Idy 0.2.2 | GitLocker.com Product

Description:

rAPIdy 0.2.2

rAPIdy
- is a minimalist web framework for those who prioritize speed and convenience.
Built upon aiohttp and pydantic,
allowing you to fully leverage the advantages of this stack.

Key features

✏️ Minimalism: Retrieve and check data effortlessly using just a single line of code
🐍 Native Python Support: Offers seamless compatibility with Python native types
📔 Pydantic Integration: Fully integrated with pydantic for robust data validation
🚀 Powered by aiohttp: Utilizes aiohttp to leverage its powerful asynchronous features
📤 Efficient Data Handling: Simplifies the extraction of basic types from incoming data in Python

Documentation

[!TIP]
Coming soon: 2024.06

Installation

[!NOTE]
pip install rapidy

Server
Quickstart
Handlers
from rapidy import web

routes = web.RouteTableDef()

@routes.post('/')
async def handler(
auth_token: str = web.Header(alias='Authorization'),
username: str = web.JsonBody(),
password: str = web.JsonBody(min_length=8),
) -> web.Response:
print({'auth_token': auth_token, 'username': username, 'password': password})
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)

[!NOTE]
rAPIdy allows you to define handlers just like in aiohttp-quickstart,
but with a key improvement:
the request parameter is no longer required for functional handlers
If you need to access the current request within a handler,
simply declare an attribute with any name of your choosing and specify its type as web.Request. rAPIdy will automatically replace this attribute with the current instance of web.Request.

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)

Middlewares
Processing an Authorization Token in Middleware
from rapidy import web
from rapidy.typedefs import HandlerType

@web.middleware
async def hello_middleware(
request: web.Request,
handler: HandlerType,
bearer_token: str = web.Header(alias='Authorization'),
) -> web.StreamResponse:
request['token'] = bearer_token
return await handler(request)

async def handler(
request: web.Request,
host: str = web.Header(alias='Host'),
username: str = web.JsonBody(),
) -> web.Response:
example_data = {'token': request['token'], 'host': host, 'username': username}
return web.json_response(example_data)

app = web.Application(middlewares=[hello_middleware])
app.add_routes([web.post('/', handler)])

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

[!IMPORTANT]
The first two attributes in a middleware are mandatory and must always represent the request and the handler respectively. These attributes are essential for the correct functioning of the middleware

Request validation
rAPIdy utilizes custom parameters to validate incoming HTTP request data, ensuring integrity and compliance with expected formats

[!TIP]
A parameter in rAPIdy represents an object that mcontains meta-information about the type of data it retrieves

A parameter in rAPIdy offers all the functionalities of pydantic.Field, and even more. This means rAPIdy-parameter support every type of validation that pydantic provides, ensuring comprehensive data integrity and conformity
from decimal import Decimal
from pydantic import BaseModel, Field
from rapidy import web

routes = web.RouteTableDef()

class Schema(BaseModel):
positive: int = Field(gt=0)
non_negative: int = Field(ge=0)
negative: int = Field(lt=0)
non_positive: int = Field(le=0)
even: int = Field(multiple_of=2)
love_for_pydantic: float = Field(allow_inf_nan=True)
short: str = Field(min_length=3)
long: str = Field(max_length=10)
regex: str = Field(pattern=r'^\d*$')
precise: Decimal = Field(max_digits=5, decimal_places=2)

@routes.get('/')
async def handler(
positive: int = web.JsonBody(gt=0),
non_negative: int = web.JsonBody(ge=0),
negative: int = web.JsonBody(lt=0),
non_positive: int = web.JsonBody(le=0),
even: int = web.JsonBody(multiple_of=2),
love_for_pydantic: float = web.JsonBody(allow_inf_nan=True),
short: str = web.JsonBody(min_length=3),
long: str = web.JsonBody(max_length=10),
regex: str = web.JsonBody(pattern=r'^\d*$'),
precise: Decimal = web.JsonBody(max_digits=5, decimal_places=2),
) -> web.Response:
return web.Response()

@routes.get('/schema')
async def handler_schema(
body: Schema = web.JsonBodySchema(),
) -> web.Response:
return web.Response()

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

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

Validation Example
from rapidy import web
from pydantic import BaseModel, Field

routes = web.RouteTableDef()

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

@routes.post('/api/{user_id}')
async def handler(
user_id: str = web.Path(),
host: str = web.Header(alias='Host'),
body: BodyRequestSchema = web.JsonBodySchema(),
) -> web.Response:
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)

Success request validation
curl -X POST \
-H "Content-Type: application/json" -d '{"username": "User", "password": "myAwesomePass"}' -v http://127.0.0.1:8080/api/1

< HTTP/1.1 200 OK ... {"data": "success"}

Failed request validation
curl -X POST \
-H "Content-Type: application/json" -d '{"username": "U", "password": "m"}' -v http://127.0.0.1:8080/api/1

< 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}
}
]
}

Types of request parameters
rAPIdy supports 3 basic types for defining incoming parameters:

Param
Schema
Raw data

Single parameter
Single parameter, used when you need to spot-retrieve incoming data.
from rapidy import web

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

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

[!NOTE]
All single parameters

Path
Header
Cookie
Query
BodyJson
FormDataBody
MultipartBody

Schema
Schema-parameter is useful when you want to extract a large amount of data.
from rapidy import web
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 handler(
path: PathRequestSchema = web.PathSchema(),
headers: HeaderRequestSchema = web.HeaderSchema(),
cookies: CookieRequestSchema = web.Cookie(),
query: QueryRequestSchema = web.QuerySchema(),
body: BodyRequestSchema = web.JsonBodySchema(),
) -> web.Response:
# write your code here
# ...
return web.Response()

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

[!NOTE]
All schema parameters

PathSchema
HeaderSchema
CookieSchema
QuerySchema
BodyJsonSchema
FormDataBodySchema
MultipartBodySchema

Raw
Use Raw-parameter when you don't need validation.
from typing import Any
from rapidy import web

async def handler(
path: dict[str, str] = web.PathRaw,
headers: dict[str, str] = web.HeaderRaw,
cookies: dict[str, str] = web.CookieRaw,
query: dict[str, str] = web.QueryRaw,
body: dict[str, Any] = web.JsonBodyRaw,
) -> web.Response:
# write your code here
# ...
return web.Response()

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

[!NOTE]
All raw parameters

PathRaw - dict[str, str]
HeaderRaw - dict[str, str]
CookieRaw - dict[str, str]
QueryRaw - dict[str, str]
BodyJsonRaw - dict[str, Any]
FormDataBodyRaw - dict[str, str] or dict[str, list[str]]
MultipartBodyRaw - dict[str, Any] or dict[str, list[Any]]
TextBody - str
BytesBody - bytes
StreamBody - aiohttp.streams.StreamReader

Combining Different Approaches
async def handler(
path_param: str = web.Path(),
headers: dict[str, str] = web.HeaderRaw(),
body: BodyRequestSchema = web.JsonBodySchema(),
) -> web.Response:

Ways to define metadata for a query parameter
There are a total of two ways to define query parameters.

Using the Annotated Auxiliary Type: Specify metadata directly within the Annotated type
Using the Default Attribute Value: Define the query parameter's default value along with its metadata

from typing import Annotated # use typing_extensions if py version == 3.8
from rapidy import web

async def handler(
param_1: Annotated[str, web.JsonBody()], # Annotated definition
param_2: str = web.JsonBody(), # Default definition
) -> web.Response:

Special attributes
Some rAPIdy-parameters include additional attributes that are not found in Pydantic, offering extended functionality beyond standard data validation
Body
Currently, only parameters of the body type include special attributes in rAPIdy

[!WARNING]
Additional attributes can only be specified for Schema-parameters and Raw-parameters.

In rAPIdy, the body_max_size attribute associated with each body parameter restricts the maximum allowable size of the request body for a specific handler
body_max_size (int) - indicating the maximum number of bytes the handler expects.
async def handler(
body: str = web.JsonBodySchema(body_max_size=10),
) -> web.Response:

async def handler(
body: str = web.JsonBodyRaw(body_max_size=10),
) -> web.Response:

Json
json_decoder (typing.Callable[[], Any]) - attribute that accepts the function to be called when decoding the body of the incoming request.
FormData
attrs_case_sensitive (bool) - attribute that tells the data extractor whether the incoming key register should be considered.
duplicated_attrs_parse_as_array (bool) - attribute that tells the data extractor what to do with duplicated keys in a query.
If duplicated_attrs_parse_as_array=True, a list will be created for each key and all values will be placed in it.

[!NOTE]
duplicated_attrs_parse_as_array flat changes the type of data that the data extractor returns.
If duplicated_attrs_parse_as_array=True, then the data
will always be of type dict[str, list[str]] (by default, formdata has the extractable type dict[str, str])

Multipart
attrs_case_sensitive (bool) - attribute that tells the data extractor whether the incoming key register should be considered.
duplicated_attrs_parse_as_array (bool) - attribute that tells the data extractor what to do with duplicated keys in a query.

[!NOTE]
duplicated_attrs_parse_as_array flat changes the type of data that the data extractor returns.
If duplicated_attrs_parse_as_array=True, then the data
will always be of type dict[str, list[Any]] (by default, multipart has the extractable type dict[str, Any])

Catch client errors
The HTTPValidationFailure exception will be raised if the data in the query is incorrect.
This error can be caught with a try/except block. The error values can be accessed using the validation_errors attribute.
This may be necessary, for example, if you need to log a customer error before giving them a response.
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(
auth_token: str = web.Header(alias='Authorization'),
) -> 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)

Default values for parameters
Some rAPIdy parameters may contain default values.
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:

[!NOTE]
Default values support some single parameters and schemes
Raw and Path parameters cannot have default values.
Some body types do not contain Raw in their name, but they are also parameters that receive raw data, such as StreamBody or TextBody.

Client

[!TIP]
Coming soon: 2024.09

OpenAPI

[!TIP]
Already in development
Coming soon: 2024.09

Mypy support
rAPIdy has its own plugin for mypy.
# example for pyproject.toml
# ...
[tool.mypy]
plugins = [
"pydantic.mypy",
"rapidy.mypy"
]
# ...

Migration from aiohttp to rAPIdy
rAPIdy neatly extends aiohttp - meaning that anything already written in aiohttp will work as before without any modifications
rAPIdy has exactly the same overridden module names as aiohttp.

[!WARNING]
rAPIdy does not override all aiohttp modules, only those that are necessary for it to work, or those that will be extended in the near future.

If the aiohttp module you are trying to override is not found in rAPIdy, don't change it, everything will work as is.

[!WARNING]
rAPIdy supports defining handlers in the same way as aiohttp-quickstart except that request is no longer a required parameter for functional handlers.
If you need to get the current request in the handler,
create an attribute with an arbitrary name and be sure to specify the web.Request type, and rAPIdy will substitute the current web.Request instance in that place.

For Developers

[!TIP]
Coming soon: 2024.06