Веб-приложение
Описание
Application — это сердце веб-сервера, отвечающее за обработку всех входящих запросов.
С его помощью можно регистрировать и управлять:
- веб-обработчиками (endpoint)
- промежуточными обработчиками (middleware)
- дочерними приложениями
- фоновыми задачами
rapidy.Rapidy является псевдонимом для rapidy.web.Application
Сущности Application
Endpoint
Endpoint — это конечная точка веб-сервиса, к которой клиентское приложение обращается для выполнения
определённых операций или получения данных, например, /api/user/get-data.
from rapidy.http import get
@get('/')
async def handler() -> dict[str, str]:
return {'hello': 'rapidy'}
Подробнее о создании endpoint смотрите в разделе Handlers.
Middleware
Middleware позволяют выполнять действия над запросом до и после его обработки веб-обработчиком.
from rapidy.http import middleware, Request, StreamResponse, Header
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)
Подробнее о создании middleware смотрите в разделе Middlewares.
Применение middleware для разных версий API.
from rapidy import web
from rapidy.typedefs import Handler
routes = web.RouteTableDef()
def get_token_data(token: str) -> ...:
...
def get_session_data(session: str) -> ...:
...
@web.middleware
async def keycloak_auth_middleware(
request: web.Request,
handler: Handler,
bearer_token: str = web.Header(alias='Authorization'),
) -> web.StreamResponse:
try:
parsed_token_data = get_token_data(bearer_token)
except Exception:
return web.HTTPUnauthorized(text='Failed to authenticate with bearer')
return await handler(request)
@web.middleware
async def cookie_session_auth_middleware(
request: web.Request,
handler: Handler,
session: str = web.Cookie(alias='UserSession'),
) -> web.StreamResponse:
try:
parsed_session_data = get_session_data(session)
except Exception:
return web.HTTPUnauthorized(text='Failed to authenticate with session')
return await handler(request)
@routes.get('/get_hello')
async def handler() -> dict[str, str]:
return {'hello': 'rapidy'}
v1_app = web.Application(middlewares=[cookie_session_auth_middleware])
v1_app.add_routes(routes)
v2_app = web.Application(middlewares=[keycloak_auth_middleware])
v2_app.add_routes(routes)
app = web.Application()
app.add_subapp('/v1', v1_app)
app.add_subapp('/v2', v2_app)
Routing
Здесь показано, как можно организовывать группы web-обработчиков.
Пример плохой практики.
Рекомендуется использовать rapidy.http.HTTPRouter или дочерние rapidy.web.Application.
HTTPRouter
Использование HTTPRouter делает код более лаконичным и удобным.
from rapidy import Rapidy # <-- rapidy.Rapidy == rapidy.web.Application
from rapidy.http import get, HTTPRouter
@get('/get_hello')
async def handler() -> dict[str, str]:
return {'hello': 'rapidy'}
v1_app = HTTPRouter('/v1', route_handlers=[handler])
rapidy = Rapidy(http_route_handlers=[v1_app])
Подробнее о HTTPRouter смотрите в разделе HTTPRouter.
Подприложение (aiohttp-style)
Создание дочерних Application:
from rapidy import web
routes = web.RouteTableDef()
@routes.get('/get_hello')
async def handler() -> dict[str, str]:
return {'hello': 'rapidy'}
v1_app = web.Application()
v1_app.add_routes(routes)
app = web.Application()
app.add_subapp('/v1', v1_app)
Использование дочерних приложений в стиле aiohttp при написании нового кода не рекомендуется.
Lifespan
Lifespan — это механизм управления жизненным циклом фоновых задач в Rapidy.
Он управляет задачами, которые должны запускаться: до или после старта сервера, а также работать постоянно.
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from rapidy import Rapidy
async def startup() -> None:
print('startup')
async def shutdown() -> None:
print('shutdown')
async def cleanup() -> None:
print('cleanup')
@asynccontextmanager
async def bg_task() -> AsyncGenerator[None, None]:
try:
print('starting background task')
yield
finally:
print('finishing background task')
rapidy = Rapidy(
on_startup=[startup],
on_shutdown=[shutdown],
on_cleanup=[cleanup],
lifespan=[bg_task()],
)
Подробнее о lifespan смотрите в разделе Lifespan.
Атрибуты Application
Дополнительные атрибуты в Rapidy
server_info_in_response
Определяет, нужно ли включать информацию о сервере в заголовке Server.
lifespan
Список фоновых задач, которые запускаются и завершаются вместе с сервером.
on_startup
Список задач, запускаемых сразу после старта приложения.
on_shutdown
Задачи, выполняемые при остановке сервера.
on_cleanup
Задачи, выполняемые после on_shutdown.
http_route_handlers
HTTP-роутеры, которые могут представлять собой как отдельные обработчики, так и группы HTTPRouter.
Атрибуты aiohttp
middlewares
Список middleware, применяемых ко всем обработчикам, включая дочерние приложения.
client_max_size
Максимальный размер запроса в байтах.
logger
Логгер для приема логов от Application.
Запуск приложения
Простой запуск
Скопируйте в файл 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)
WSGI запуск (Gunicorn)
Установите 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:app обращается к:
main: файлmain.py(модуль Python).rapidy: объект, созданный внутри файлаmain.pyв строкеrapidy = Rapidy().--reload: перезапускает сервер после изменения кода. Используйте только для разработки.