Маршрутизация и создание HTTP-обработчиков

Веб-обработчики (handlers) предназначены для обработки входящих HTTP-запросов.

В Rapidy маршрутизация и создание обработчиков тесно связаны: маршрутизация определяет, какой обработчик будет вызван в ответ на запрос.

Определение маршрутов

Маршрут — это строка URL, по которой вызывается обработчик.

Всего существуют два вида маршрутов - статические и динамические.

Разница между статическим и динамическим роутингом

Тип маршрута	Пример URL	Описание
Статический	/about	URL фиксирован.
Динамический	/users/{user_id}	URL меняется в зависимости от параметров

Rapidy поддерживает несколько способов определения маршрутов, аналогичных aiohttp, подробнее о них ниже в разделе Создание и регистрация обработчиков HTTP-запросов.

Подробнее про обработчики aiohttp можно узнать здесь.

Статические маршруты

Статический HTTP-роутинг — это маршрутизация, где путь (URL) заранее известен и не меняется динамически. Это означает, что каждый запрос с определённым маршрутом всегда приводит к одному и тому же обработчику.

Простой статический маршрут

from rapidy.http import get

@get('/hello_rapidy')
async def handler() -> str:
    return 'Hello Rapidy!'

Этот маршрут всегда доступен по GET /hello и возвращает один и тот же ответ.

curl http://localhost:8000/hello_rapidy

Таким же образом вы можете определять и другие методы, такие как get, post, put, delete и так далее.

Динамические маршруты

Динамический роутинг позволяет задавать маршруты, которые принимают переменные параметры. Это полезно, когда нужно работать с различными сущностями (например, user_id, post_id и т.д.), передавая их в URL.

В примерах ниже будет использоваться PathParam, который необходим для извлечения параметров-путей. Подробнее о нем можно прочитать здесь.

Простой динамический маршрут

Допустим, у нас есть API для получения информации о пользователе по его user_id:

from rapidy.http import get, PathParam

@get('/users/{user_id}')
async def handler(user_id: int = PathParam()) -> dict[str, int]:
    return {'user_id': user_id}

Как работает этот маршрут?

user_id — динамический параметр, который передаётся в URL.
Rapidy автоматически преобразует его в int (если передать строку, API вернёт ошибку 422).

Пример запроса:

curl http://localhost:8000/users/123

Ответ:

{"user_id": 123}

Динамические маршруты с несколькими параметрами

Можно добавить несколько динамических параметров:

from rapidy.http import get, PathParam

@get('/posts/{post_id}/comments/{comment_id}')
async def handler(
    post_id: int = PathParam(),
    comment_id: int = PathParam(),
) -> dict[str, int]:
    return {'post_id': post_id, 'comment_id': comment_id}

Теперь запрос GET /posts/10/comments/5 вернёт:

{"post_id": 10, "comment_id": 5}

Группировка маршрутов

Если у вас много маршрутов, можно использовать один из подходов для группировки HTTP-запросов.

Рекомендуется придерживаться только одного подхода в рамках проекта.

HTTPRouter

Rapidy предлагает использовать объект HTTPRouter для группировки запросов.

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

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

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

@get('/healthcheck')  # /healthcheck
async def healthcheck() -> str:
    return 'ok'

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

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

rapidy = Rapidy(http_route_handlers=[healthcheck, api_router])

HTTPRouter умеет больше!

HTTPRouter также имеет ряд атрибутов расширяющих его возможности, такие как обработка middleware, управление фоновыми задачами и тд.

Также вы можете создавать вложенные HTTPRouter.

Подробнее про HTTPRouter можно прочитать здесь

Создание и регистрация обработчиков HTTP-запросов

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

Простейший способ создания обработчика:

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),
    ]
)

Добавление обработчика через router приложения (в стиле aiohttp)

from rapidy import Rapidy

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

rapidy = Rapidy()
rapidy.router.add_post('/', handler)

Поддерживаемые методы соответствуют HTTP-методам с префиксом add_.

add_get
add_post
add_put
add_patch
add_delete

Исключение — view.

add_view

Добавление обработчика с декоратором через RouteTableDef (в стиле aiohttp)

from rapidy import web, Rapidy

routes = web.RouteTableDef()

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

rapidy = Rapidy()
rapidy.add_routes(routes)

Добавление обработчика без декоратора через rapidy.web (в стиле aiohttp)

from rapidy import web, Rapidy

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

rapidy = Rapidy()
rapidy.add_routes([web.post('/', 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),
    ]
)

Использование `View` (aiohttp style)

Добавление обработчика через router приложения (в стиле aiohttp)

from rapidy import Rapidy
from rapidy.web import View

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

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

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

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

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

rapidy = Rapidy()
rapidy.router.add_view('/', Handler)

Добавление обработчика через router с разными путями (в стиле aiohttp)

from rapidy import Rapidy
from rapidy.web import View, PathParam

class Handler(View):
    async def get(self, user_id: str = PathParam()) -> dict[str, str]:
        return {'hello': 'rapidy'}

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

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

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

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

rapidy = Rapidy()
rapidy.router.add_get('/{user_id}', Handler)
rapidy.router.add_view('/', Handler)

Добавление обработчика с декоратором через RouteTableDef (в стиле aiohttp)

from rapidy import web

routes = web.RouteTableDef()

@routes.view('/')
class Handler(web.View):
    async def get(self) -> dict[str, str]:
        return {'hello': 'rapidy'}

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

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

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

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

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

Добавление обработчика с декоратором через RouteTableDef с разными путями (в стиле aiohttp)

from rapidy import web

routes = web.RouteTableDef()

@routes.view('/')
class Handler(web.View):
    @routes.get('/{user_id}')
    async def get(self, user_id: str = web.PathParam()) -> dict[str, str]:
        return {'hello': 'rapidy'}

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

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

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

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

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

Добавление обработчика без декоратора через rapidy.web (в стиле aiohttp)

from rapidy import web

routes = web.RouteTableDef()

class Handler(web.View):
    async def get(self) -> dict[str, str]:
        return {'hello': 'rapidy'}

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

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

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

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

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

Добавление обработчика без декоратора через rapidy.web с разными путями (в стиле aiohttp)

from rapidy import web

routes = web.RouteTableDef()

class Handler(web.View):
    async def get(self, user_id: str = web.PathParam()) -> dict[str, str]:
        return {'hello': 'rapidy'}

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

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

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

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

rapidy = web.Application()
rapidy.add_routes([
    web.get('/{user_id}', Handler),
    web.view('/', Handler),
])

Атрибуты обработчиков

Атрибуты позволяют управлять поведением обработчиков и ответами.

Атрибуты автоматически применяются к ответам обработчиков, если обработчик возвращает что угодно кроме Response (не относится к атрибутам path и allow_head для метода get).

Атрибуты применяются к ответам.

Атрибут response_content_type будет применяться к каждому ответу обработчика, поскольку обработчик возвращает python объект.

from rapidy.http import get, ContentType

@get('/', response_content_type=ContentType.text_plain)
async def handler() -> str:
    return 'Hello Rapidy!'

Атрибуты не применяются к ответам.

Атрибут response_content_type не будет применяться к ответу обработчика, поскольку обработчик возвращает низкоуровневый Response объект.

from rapidy.http import get, ContentType, Response

@get('/', response_content_type=ContentType.text_plain)
async def handler() -> Response:
    return Response('Hello Rapidy!')

Все способы создания обработчиков поддерживают одинаковые атрибуты для управления веб-запросом.

Основные атрибуты (применяются всегда)

path

path: str — маршрут обработчика на сервере.

@get(
    '/',
)
async def handler() -> str:
    return 'ok'

allow_head

allow_head: bool = True — если равен True (по умолчанию), то добавляется маршрут для метода head с тем же обработчиком, что и для get.

@get(
    '/',
    allow_head=True,
)
async def handler() -> str:
    return 'ok'

Аттрибут может быть применен только к методу get.

Валидация ответа

response_validate

response_validate: bool = True — проверять ли ответ обработчика.

@get(
    '/',
    response_validate=False,
)
async def handler() -> str:  # <-- `str` will be ignored
    return {'hello': 'rapidy'}

response_type

response_type: Type[Any] | None = ... — определяет тип ответа (заменяет аннотацию возврата).

@get(
    '/',
    response_type=dict[str, str],  # <-- `dict[str, str]` will be used for validation
)
async def handler() -> str:  # <-- `str` will be ignored
    return {'hello': 'rapidy'}

Этот флаг добавляет гибкость в сериализацию и валидацию, но используется редко.

Управление заголовками и кодировкой

response_content_type

response_content_type: str = 'application/json' — аттрибут позволяющий управлять заголовком Content-Type.

Заголовок Content-Type сообщает клиенту (браузеру, API-клиенту, другому серверу), какой тип данных содержится в теле HTTP-ответа.

from rapidy.enums import ContentType

@get(
    '/',
    response_content_type=ContentType.text_plain,
)
async def handler() -> str:
    return 'hello, rapidy!'

Если указан content_type, переданные данные будут преобразованы в соответствии с ним.

Если content_type не указан - content_type будет определен автоматически в зависимости от типа данных которое отдает сервер.

content_type="application/json

content_type="application/json" — данные преобразуются в JSON с использованием jsonify(dumps=True) и кодируются в соответствии с response_charset.

from rapidy.http import get, ContentType

@get(
    '/',
    content_type=ContentType.json,
)
async def handler() -> dict[str, str]:
    return {'hello': 'rapidy!'}  # {"hello": "rapidy!"}

Если переданный объект является строкой Response(body="string"), то строка, согласно стандарту JSON, будет экранирована дважды:

from rapidy.http import get, ContentType

@get(
    '/',
    content_type=ContentType.json,
)
async def handler() -> str:
    return 'hello rapidy!'  # "'hello rapidy!'"

content_type="text/*

content_type="text/*" (любой текстовый тип: text/plain, text/html и т. д.) - если данные имеют тип str, они отправляются без изменений. В противном случае они преобразуются в строку через jsonify(dumps=False).

from rapidy.http import get, ContentType

@get(
    '/',
    content_type=ContentType.text_any,
)
async def handler() -> str:
    return 'hello rapidy!'  # "hello rapidy!"

Если после jsonify(dumps=False) объект все еще не является строкой, он дополнительно преобразуется с помощью response_json_encoder.

content_type - любой другой MIME-type.

Если данные имеют тип bytes, они отправляются без изменений. В противном случае они преобразуются в строку с использованием jsonify(dumps=True) и кодируются в соответствии с response_json_encoder.

Если content_type не указан, он устанавливается автоматически:

body: dict | BaseModel | dataclass → content_type="application/json"

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

async def handler() -> SomeModel:
    return SomeModel(hello="rapidy")  # `SomeModel` inherits from `pydantic.BaseModel`

async def handler() -> str:
    return 'string'

async def handler() -> str:
    return SomeEnum.string

async def handler() -> int:
    return 1

async def handler() -> float:
    return 1.0

async def handler() -> Decimal:
    return Decimal("1.0")

async def handler() -> bool:
    return True

body: Any → content_type="application/octet-stream"

async def handler() -> bytes:
    return b'bytes'

async def handler() -> AnotherType:
    return AnotherType()

response_charset

response_charset: str = 'utf-8' — кодировка, используемая для кодирования данных ответа.

from rapidy.enums import Charset

@get(
    '/',
    response_charset=Charset.utf8,
)
async def handler() -> str:
    return 'hello, rapidy!'

response_json_encoder

response_json_encoder: Callable = json.dumps — функция, принимающая объект и возвращающая его JSON-представление.

Автоматически применяется к любому python объекту после валидации его через pydantic.

from typing import Any

def custom_encoder(obj: Any) -> str:
    ...

@get(
    '/',
    response_json_encoder=custom_encoder,  # Converts the obtained string above into a JSON object using the `custom_encoder` function
)
async def handler() -> dict[str, str]:
    return {'hello': 'rapidy!'}  # will be converted to a string by Rapidy's internal tools

Сжатие данных

response_zlib_executor

response_zlib_executor: concurrent.futures.Executor | None = None — функция сжатия zlib.

from concurrent.futures import Executor

class SomeExecutor(Executor):
    ...

@get(
    '/',
    response_zlib_executor=SomeExecutor,
)
async def handler() -> str:
    return 'hello, rapidy!'

Подробнее о zlib_executor

zlib_executor — механизм aiohttp. Подробнее здесь.

response_zlib_executor

response_zlib_executor_size: int | None = None — размер тела в байтах для включения сжатия.

@get(
    '/',
    response_zlib_executor_size=1024,
)
async def handler() -> str:
    return 'hello, rapidy!'

Управление полями Pydantic

response_include_fields

response_include_fields**: set[str] | dict[str, Any] | None = None — параметр include из Pydantic, указывающий, какие поля включать.

from pydantic import BaseModel, Field

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    another_value: str = Field('another_data', alias='someAnotherValue')

@get(
    '/',
    response_include_fields={'value'},
)
async def handler() -> Result:
    return Result()  # {'someValue': 'data'}

response_exclude_fields

response_exclude_fields: set[str] | dict[str, Any] | None — список полей для исключения.

from pydantic import BaseModel, Field

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    another_value: str = Field('another_data', alias='someAnotherValue')

@get(
    '/',
    response_exclude_fields={'value'},
)
async def handler() -> Result:
    return Result()  # {"someValue": "data"}

response_by_alias

response_by_alias: bool = True — использовать ли псевдонимы Pydantic.

from pydantic import BaseModel, Field

class Result(BaseModel):
    value: str = Field('data', alias='someValue')

@get(
    '/',
    response_by_alias=True,  # <-- default
)
async def handler() -> Result:
    return Result()  # {"someValue": "data"}

...

@get(
    '/',
    response_by_alias=False,
)
async def handler() -> Result:
    return Result()  # {"value": "data"}

response_exclude_unset

response_exclude_unset: bool = False — исключать ли значения по умолчанию.

from pydantic import BaseModel, Field

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    another_value: str = Field('another_data', alias='someAnotherValue')

@get(
    '/',
    exclude_unset=False,  # <-- default
)
async def handler() -> Result:
    Result(someAnotherValue='new_data')  # {"someValue": "data", "someAnotherValue": "new_data"}

...

@get(
    '/',
    exclude_unset=True,
)
async def handler() -> Result:
    return Result(someAnotherValue='new_data')  # {"someAnotherValue": "new_data"}

response_exclude_defaults

response_exclude_defaults: bool = False — исключать ли явно заданные значения, если они совпадают с дефолтными.

from pydantic import BaseModel, Field

class Result(BaseModel):
    value: str = Field('data', alias='someValue')

@get(
    '/',
    exclude_defaults=False,  # <-- default
)
async def handler() -> Result:
    return Result()  # {"value": "data"}

...

@get(
    '/',
    exclude_defaults=True,
)
async def handler() -> Result:
    return Result()  # {}

response_exclude_none

response_exclude_none: bool = False — исключать ли None-значения.

from pydantic import BaseModel, Field

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    none_value: None = None

@get(
    '/',
    exclude_none=False,  # <-- default
)
async def handler() -> Result:
    return Result()  # {"someValue": "data", "none_value": null}

...

@get(
    '/',
    exclude_none=True,
)
async def handler() -> Result:
    return Result()  # {"someValue": "data"}

response_custom_encoder

response_custom_encoder: Callable | None = None — параметр custom_encoder из Pydantic, позволяющий задать пользовательский кодировщик.

Маршрутизация и создание HTTP-обработчиков

Определение маршрутов

Статические маршруты

Динамические маршруты

Простой динамический маршрут

Динамические маршруты с несколькими параметрами

Группировка маршрутов

HTTPRouter

Создание и регистрация обработчиков HTTP-запросов

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

Примеры регистрации обработчиков

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

Примеры регистрации классовых обработчиков

Использование View (aiohttp style)

Атрибуты обработчиков

Основные атрибуты (применяются всегда)

path

allow_head

Валидация ответа

response_validate

response_type

Управление заголовками и кодировкой

response_content_type

response_charset

response_json_encoder

Сжатие данных

response_zlib_executor

response_zlib_executor

Управление полями Pydantic

response_include_fields

response_exclude_fields

response_by_alias

response_exclude_unset

response_exclude_defaults

response_exclude_none

response_custom_encoder

Использование `View` (aiohttp style)