Быстрый старт

Создание простых обработчиков

rAPIdy базируется на aiohttp и поддерживает все варианты определения обработчиков.
Ниже будут показаны лишь основные.

Все способы определения можно посмотреть здесь

Функциональные обработчики

from rapidy import Rapidy
from rapidy.http import post

@post('/')
async def handler() -> dict[str, str]:
    return {'hello': 'rapidy'}

rapidy = Rapidy(http_route_handlers=[handler])

Регистрация без декоратора

from rapidy import Rapidy
from rapidy.http import post

async def handler() -> dict[str, str]:
    return {'hello': 'rapidy'}

rapidy = Rapidy(
    http_route_handlers=[
        post.reg('/', handler),
    ]
)

---

Классовые обработчики

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

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

    @get()
    async def get_all_users(self) -> list[dict[str, str]]:
        return [{'name': 'John'}, {'name': 'Felix'}]

    @post()
    async def create_user(self) -> str:
        return 'ok'

    @put()
    async def update_user(self) -> str:
        return 'ok'

    @patch()
    async def patch_user(self) -> str:
        return 'ok'

    @delete()
    async def delete_user(self) -> str:
        return 'ok'

rapidy = Rapidy(http_route_handlers=[UserController])

Регистрация без декоратора

from rapidy import Rapidy
from rapidy.http import PathParam, controller, get

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

    @get()
    async def get_all_users(self) -> list[dict[str, str]]:
        return [{'name': 'John'}, {'name': 'Felix'}]

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

Middleware

Middleware — это промежуточные компоненты, которые обрабатывают запросы и ответы на уровне веб-фреймворка.

Они позволяют выполнять дополнительные действия до и/или после обработки запроса основным обработчиком.

from rapidy import Rapidy
from rapidy.http import Request, StreamResponse, get, middleware
from rapidy.typedefs import CallNext

@middleware
async def hello_middleware(request: Request, call_next: CallNext) -> StreamResponse:
    request['data'] = {'hello': 'rapidy'}
    return await call_next(request)

@get('/')
async def handler(request: Request) -> dict[str, str]:
    return request['data']

rapidy = Rapidy(
    http_route_handlers=[handler],
    middlewares=[hello_middleware],
)

Примечание

В middleware-обработчике первым аргументом всегда должен быть Request, а вторым — call_next.

HTTPRoute

HTTPRoute позволяет регистрировать группы обработчиков и играет ключевую роль в маршрутизации (routing), помогая направлять запросы к нужным обработчикам в зависимости от HTTP-метода, пути, параметров и других условий.

HTTPRoute регистрируется точно так же, как и любой HTTP-обработчик.

from rapidy import Rapidy
from rapidy.http import HTTPRouter, controller, get

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

@controller('/hello_controller')
class HelloController:
    @get()
    async def get_hello(self) -> dict[str, str]:
        return {'hello': 'rapidy'}

api_router = HTTPRouter('/api', [hello_handler, HelloController])

rapidy = Rapidy(http_route_handlers=[api_router])

Регистрация обработчиков в роутере без декоратора

from rapidy import Rapidy
from rapidy.http import HTTPRouter, controller, get

async def hello_handler() -> dict[str, str]:
    return {'hello': 'rapidy'}

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

api_router = HTTPRouter(
    '/api',
    [
        get.reg('/hello', hello_handler),
        controller.reg('/hello_controller', HelloController),
    ]
)

rapidy = Rapidy(http_route_handlers=[api_router])

Полный пример регистрации роутера и обработчиков

from rapidy import Rapidy
from rapidy.http import controller, get, HTTPRouter

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

@controller('/hello_controller')
class HelloController:
    @get()
    async def get(self) -> dict[str, str]:
        return {'hello': 'rapidy'}

@get('/hi')
async def hi_handler() -> dict[str, str]:
    return {'hi': 'rapidy'}

@controller('/hi_controller')
class HiController:
    @get()
    async def get(self) -> dict[str, str]:
        return {'hi': 'rapidy'}

hello_api_router = HTTPRouter('/hello_api', [hello_handler, HelloController])

rapidy = Rapidy(
    http_route_handlers=[
        hello_api_router,
        hi_handler,
        HiController,
    ]
)

Пример простой валидации

from pydantic import BaseModel
from rapidy import Rapidy
from rapidy.http import 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
    password: str

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('/')
async def handler(body: RequestBody = Body()) -> ResponseBody:
    return ResponseBody()

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

Запуск веб-сервера

Простой запуск

Скопируйте код в файл main.py.

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

@post('/')
async def handler() -> dict[str, str]:
    return {'hello': 'rapidy'}

rapidy = Rapidy()
rapidy.add_http_router(handler)

if __name__ == '__main__':
    run_app(rapidy)

Запустите сервер в режиме реального времени:

python3 main.py

Также вы можете указать различные параметры запуска, например, host или port.

run_app(app, host='0.0.0.0', port=8080)

WSGI-запуск (Gunicorn)

Gunicorn — это Python WSGI HTTP-сервер для UNIX.

Что такое WSGI

WSGI (Web Server Gateway Interface — интерфейс шлюза Web-сервера) — это простой и универсальный интерфейс взаимодействия между Web-сервером и Web-приложением, впервые описанный в PEP-333.

Подробнее про gunicorn.org

Добавьте Gunicorn в ваш проект:

pip install gunicorn

Скопируйте код в файл main.py.

from rapidy import Rapidy
from rapidy.http import post

@post('/')
async def handler() -> dict[str, str]:
    return {'hello': 'rapidy'}

rapidy = Rapidy()
rapidy.add_http_router(handler)

Запустите команду в консоли:

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

Команда gunicorn main:rapidy обращается к:

• main: файл main.py (модуль Python).
• rapidy: объект, созданный внутри файла main.py в строке rapidy = Rapidy().
• --reload: перезапускает сервер после изменения кода. Используйте только для разработки.