rAPIdy - a fast, lightweight, and modern asynchronous web framework powered by aiohttp and pydantic.

⭐️ If you like rAPIdy, consider starring the project on GitHub — it helps a lot!

rAPIdy is designed for developers who need a fast, async-first web framework that combines the performance of aiohttp with the simplicity and modern features of frameworks like FastAPI.

Simple rAPIdy server:

from rapidy import Rapidy
from rapidy.http import get

@get("/")
async def hello() -> dict[str, str]:
    return {"message": "Hello, rAPIdy!"}

app = Rapidy(http_route_handlers=[hello])

• Fast & Lightweight – Minimal overhead, built on aiohttp
• Async-First – Fully asynchronous by design
• Built-in Validation – Uses pydantic for request validation
• Simple & Flexible – Supports both rAPIdy-style handler definitions and traditional aiohttp function-based and class-based routing
• Middleware Support – Easily extend functionality with middleware, including built-in validation for HTTP parameters (headers, cookies, and body).

Install rAPIdy via pip:

pip install rapidy

Documentation: https://rapidy.dev

Updates Channel: https://t.me/rapidy_dev

Simple rAPIdy server:

Copy the following code into a file named main.py:

from rapidy import Rapidy
from rapidy.http import get

@get("/")
async def hello() -> dict[str, str]:
    return {"message": "Hello, rAPIdy!"}

rapidy = Rapidy(http_route_handlers=[hello])

There are several ways to start the server:

• Using run_app
• Using WSGI (gunicorn)

from rapidy import Rapidy, run_app
from rapidy.http import get

@get("/")
async def hello() -> dict[str, str]:
    return {"message": "Hello, rAPIdy!"}

rapidy = Rapidy(http_route_handlers=[hello])

if __name__ == '__main__':
    run_app(rapidy, host="0.0.0.0", port=8000)

python3 main.py

pip install gunicorn

from rapidy import Rapidy
from rapidy.http import get

@get("/")
async def hello() -> dict[str, str]:
    return {"message": "Hello, rAPIdy!"}

rapidy = Rapidy(http_route_handlers=[hello])

gunicorn main:rapidy --bind localhost:8000 --reload --worker-class aiohttp.GunicornWebWorker

Define multiple routes with function-based or class-based views:

Functional Handlers:

from rapidy.http import post, Body

@post("/items")
async def create_item(item: dict[str, str] = Body()) -> dict[str, str]:
    return item

from rapidy import Rapidy
from rapidy.http import post, Body

async def create_item(item: dict[str, str] = Body()) -> dict[str, str]:
    return item

rapidy = Rapidy()
rapidy.add_http_routers([post.reg('/items', create_item)])

Class-Based Handlers:

All methods decorated with @get, @post, etc., are automatically registered as sub-routes.

from rapidy.http import controller, get, post, put, patch, delete, PathParam, Body

@controller('/')
class ItemController:
    @get('/{item_id}')
    async def get_by_id(self, item_id: str = PathParam()) -> dict[str, str]:
        return {'hello': 'rapidy'}

    @get()
    async def get_all(self) -> list[dict[str, str]]:
        return [{'hello': 'rapidy'}, {'hello': 'rapidy'}]

    @post()
    async def post(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

    @put()
    async def put(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

    @patch()
    async def patch(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

    @delete()
    async def delete(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

from rapidy import Rapidy
from rapidy.http import PathParam, get, post, put, patch, delete, controller, Body

class ItemController:
    @get('/{item_id}')
    async def get_by_id(self, item_id: str = PathParam()) -> dict[str, str]:
        return {'hello': 'rapidy'}

    @get()
    async def get_all(self) -> list[dict[str, str]]:
        return [{'hello': 'rapidy'}, {'hello': 'rapidy'}]

    @post()
    async def post(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

    @put()
    async def put(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

    @patch()
    async def patch(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

    @delete()
    async def delete(self, item: dict[str, str] = Body()) -> dict[str, str]:
        return item

rapidy = Rapidy(
    http_route_handlers=[controller.reg('/', ItemController)],
)

rAPIdy uses pydantic not only to validate request data but also for response data serialization. This ensures data consistency and type safety throughout the entire request-response cycle:

from rapidy.http import post, Body
from pydantic import BaseModel

class ItemSchema(BaseModel):
    name: str
    price: float

@post("/items")
async def create_item(data: ItemSchema = Body()) -> ItemSchema:
    return data

If validation fails, a 422 Unprocessable Entity response with error details is automatically returned.

from pydantic import BaseModel, Field
from rapidy import Rapidy
from rapidy.http import PathParam, Body, Request, Header, StreamResponse, middleware, post
from rapidy.typedefs import CallNext

TOKEN_REGEXP = '^[Bb]earer (?P<token>[A-Za-z0-9-_=]*)'

class RequestBody(BaseModel):
    username: str = Field(min_length=2, max_length=50)
    password: str = Field(min_length=2, max_length=50)

class ResponseBody(BaseModel):
    hello: str = 'rapidy'

@middleware
async def get_bearer_middleware(
        request: Request,
        call_next: CallNext,
        bearer_token: str = Header(alias='Authorization', pattern=TOKEN_REGEXP),
) -> StreamResponse:
    # process token here ...
    return await call_next(request)

@post('/{user_id}')
async def handler(
        user_id: str = PathParam(),
        body: RequestBody = Body(),
) -> str:
    return 'success'

app = Rapidy(
    middlewares=[get_bearer_middleware],
    http_route_handlers=[handler],
)

curl -X POST \
-H "Content-Type: application/json" \
-H "Authorization: Bearer my-token" \
-d '{"username": "Username", "password": "Password"}' \
-v http://127.0.0.1:8080/1

< HTTP/1.1 200 OK ...
success

curl -X POST \
-H "Content-Type: application/json" \
-H "Authorization: Bearer my-token" \
-d '{"username": "U", "password": "P"}' \
-v http://127.0.0.1:8080/1

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

Easily add powerful middleware for authentication, logging, and more. rAPIdy allows you to validate HTTP parameters (headers, cookies, body) directly within middleware for enhanced flexibility:

Middleware can be used for authentication, logging, or other cross-cutting concerns.

from rapidy import Rapidy
from rapidy.http import PathParam, Request, Header, StreamResponse, middleware, post
from rapidy.typedefs import CallNext

TOKEN_REGEXP = '^[Bb]earer (?P<token>[A-Za-z0-9-_=]*)'

@middleware
async def get_bearer_middleware(
    request: Request,
    call_next: CallNext,
    bearer_token: str = Header(alias='Authorization', pattern=TOKEN_REGEXP),
) -> StreamResponse:
    # process token here ...
    return await call_next(request)

@post('/{user_id}')
async def handler(user_id: str = PathParam()) -> str:
    return 'success'

app = Rapidy(
    middlewares=[get_bearer_middleware],
    http_route_handlers=[handler],
)

Rapidy uses the dishka library as its built-in Dependency Injection (DI) mechanism.

We aimed to choose a DI library aligned with the philosophy of Rapidy: simplicity, speed, transparency, and scalability. dishka perfectly fits these principles, offering developers a powerful tool without unnecessary complexity.

In Rapidy, dishka is available out-of-the-box — no additional setup required.

from rapidy import Rapidy
from rapidy.http import Request, StreamResponse, get, middleware
from rapidy.typedefs import CallNext
from rapidy.depends import provide, Provider, Scope, FromDI

class FooProvider(Provider):
    @provide(scope=Scope.REQUEST)
    async def some_obj(self) -> int:
        return 1

@middleware
async def some_middleware(
    request: Request,
    call_next: CallNext,
    some_obj: FromDI[int],
) -> StreamResponse:
    print({"value": some_obj})
    return await call_next(request)

@get('/')
async def handler(some_obj: FromDI[int]) -> dict:
    return {"value": some_obj}

app = Rapidy(
    middlewares=[some_middleware],
    http_route_handlers=[handler],
    di_providers=[FooProvider()],
)

To gain a deeper understanding of how the DI mechanism works, refer to the documentation for Rapidy and dishka.

Lifespan is a lifecycle manager for background tasks within Rapidy.

Although aiohttp supports the background tasks feature, rapidy does it more conveniently.

Lifespan manages tasks that should be started: before or after server startup, or should always run.

There are several ways to start tasks: on_startup, on_shutdown, on_cleanup, lifespan.

on_startup - tasks that will be executed in the event loop along with the application's request handlers immediately after the application starts.

from rapidy import Rapidy

def startup() -> None:
    print('startup')

rapidy = Rapidy(on_startup=[startup])

from rapidy import Rapidy

def startup(rapidy: Rapidy) -> None:
    print(f'startup, application: {rapidy}')

rapidy = Rapidy(on_startup=[startup])

from rapidy import Rapidy

async def async_startup() -> None:
    print('async_startup')

rapidy = Rapidy(on_startup=[async_startup])

from rapidy import Rapidy

async def async_startup(rapidy: Rapidy) -> None:
    print(f'async_startup, application: {rapidy}')

rapidy = Rapidy(on_startup=[async_startup])

Adding on_startup to an already created Application object.

rapidy = Rapidy()
rapidy.lifespan.on_startup.append(startup)

on_shutdown - tasks that will be executed after the server stops.

from rapidy import Rapidy

def shutdown() -> None:
    print('shutdown')

rapidy = Rapidy(on_shutdown=[shutdown])

from rapidy import Rapidy

def shutdown(rapidy: Rapidy) -> None:
    print(f'shutdown, application: {rapidy}')

rapidy = Rapidy(on_shutdown=[shutdown])

from rapidy import Rapidy

async def async_shutdown() -> None:
    print('async_shutdown')

rapidy = Rapidy(on_shutdown=[async_shutdown])

from rapidy import Rapidy

async def async_shutdown(rapidy: Rapidy) -> None:
    print(f'async_shutdown, application: {rapidy}')

rapidy = Rapidy(on_shutdown=[async_shutdown])

Adding on_shutdown to an already created Application object.

rapidy = Rapidy()
rapidy.lifespan.on_shutdown.append(shutdown)

on_cleanup - tasks that will be executed after the server stops and all on_shutdown tasks are completed.

from rapidy import Rapidy

def cleanup() -> None:
    print('cleanup')

rapidy = Rapidy(on_cleanup=[cleanup])

from rapidy import Rapidy

def cleanup(rapidy: Rapidy) -> None:
    print(f'cleanup, application: {rapidy}')

rapidy = Rapidy(on_cleanup=[cleanup])

from rapidy import Rapidy

async def async_cleanup() -> None:
    print('async_cleanup')

rapidy = Rapidy(on_cleanup=[async_cleanup])

from rapidy import Rapidy

async def async_cleanup(rapidy: Rapidy) -> None:
    print(f'async_cleanup, application: {rapidy}')

rapidy = Rapidy(on_cleanup=[async_cleanup])

Adding on_cleanup to an already created Application object.

rapidy = Rapidy()
rapidy.lifespan.on_cleanup.append(cleanup)

lifespan - manages background tasks.

from contextlib import asynccontextmanager
from typing import AsyncGenerator
from rapidy import Rapidy

@asynccontextmanager
async def bg_task() -> AsyncGenerator[None, None]:
    try:
        print('starting background task')
        yield
    finally:
        print('finishing background task')

rapidy = Rapidy(
    lifespan=[bg_task()],
)

from contextlib import asynccontextmanager
from typing import AsyncGenerator
from rapidy import Rapidy

@asynccontextmanager
async def bg_task_with_app(rapidy: Rapidy) -> AsyncGenerator[None, None]:
    try:
        print('starting background task')
        yield
    finally:
        print('finishing background task')

rapidy = Rapidy(
    lifespan=[bg_task_with_app],
)

Adding lifespan to an already created Application object.

rapidy = Rapidy()
rapidy.lifespan.append(bg_task())

You can use pytest and pytest-aiohttp for testing your rAPIdy application:

Install pytest and pytest-aiohttp

pip install pytest pytest-aiohttp

Example of a simple test:

from rapidy import Rapidy
from rapidy.http import get
from pytest_aiohttp.plugin import AiohttpClient

@get('/')
async def hello() -> dict[str, str]:
    return {'message': 'Hello, rAPIdy!'}

async def test_hello(aiohttp_client: AiohttpClient) -> None:
    app = Rapidy(http_route_handlers=[hello])

    client = await aiohttp_client(app)

    resp = await client.get('/')
    assert resp.status == 200
    assert await resp.json() == {'message': 'Hello, rAPIdy!'}

rAPIdy is built on top of aiohttp, offering a familiar development experience with powerful enhancements:

• Full Compatibility with aiohttp Syntax – rAPIdy fully supports the definition of HTTP handlers just like in aiohttp, offering the same capabilities. For more details, refer to the rAPIdy documentation.
• Cleaner Routing Syntax – No need for web.RouteTableDef, making route definitions more concise and readable.
• Significantly Reduced Boilerplate Code – rAPIdy minimizes the amount of code required compared to aiohttp, allowing developers to focus on business logic rather than repetitive setup.
• Built-in Request Validation and Response Serialization – Powered by pydantic, rAPIdy automatically validates incoming requests and serializes responses, ensuring data consistency and reducing potential errors.
• Powerful Middlewares – First-class support for middleware and easy-to-use dependency injection out of the box.
• Lifespan support.

rAPIdy includes its own plugin for mypy.

The rapidy.mypy plugin helps with type checking in code that uses rAPIdy.

Add the auxiliary configuration to your mypy configuration file.

mypy.ini

; Example configuration for mypy.ini
; ...
[tool.mypy]
plugins =
    pydantic.mypy
    rapidy.mypy
; ...

pyproject.toml

# Example configuration for pyproject.toml
# ...
[tool.mypy]
plugins = [
    "pydantic.mypy",
    "rapidy.mypy"
]
# ...

We're actively improving rAPIdy to make it more powerful and efficient. Stay up-to-date with our latest milestones and future plans by checking out the detailed roadmap below.

You can find the full and detailed roadmap on our GitHub page: ROADMAP.md

Want to improve rAPIdy? We welcome contributions! 🚀

• Report Issues: GitHub Issues
• Pull Requests: Fork & create PRs!
• Contribution Guide: CONTRIBUTING.md

rAPIdy is licensed under the MIT License. See LICENSE for details.

Start building fast, reliable, and modern APIs with rAPIdy today! 🚀