Application

Description

Application is the core of the web server responsible for handling all incoming requests.

It allows you to register and manage:

Web handlers (endpoint)
Middleware
Sub-applications
Background tasks

rapidy.Rapidy is an alias for rapidy.web.Application

Application Entities

Endpoint

An endpoint is a final point of a web service that a client application interacts with to perform specific operations or retrieve data, such as /api/user/get-data.

from rapidy.http import get

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

For more details on creating an endpoint, see the Handlers section.

Middleware

Middleware allows you to perform actions on a request before and after it is processed by a web handler.

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)

For more details on creating middleware, see the Middlewares section.

Applying middleware for different API versions.

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

This section demonstrates how to organize groups of web handlers.

Example of bad practice.

@routes.get('/api/v1/get_hello')  # <-- bad practice
async def handler() -> ...:

It is recommended to use rapidy.http.HTTPRouter or child rapidy.web.Application instead.

HTTPRouter

Using HTTPRouter makes the code more concise and convenient.

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

curl -X GET http://127.0.0.1:8080/v1/get_hello

For more details on HTTPRouter, see the HTTPRouter section.

Sub-application (aiohttp-style)

Creating child 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)

curl -X GET http://127.0.0.1:8080/v1/get_hello

Using child applications in the aiohttp style for new code is not recommended.

Lifespan

Lifespan is a mechanism for managing the lifecycle of background tasks in Rapidy.

It controls tasks that should be executed: before or after the server starts, as well as tasks that should run continuously.

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

For more details on lifespan, see the Lifespan section.

Application Attributes

Additional Attributes in `Rapidy`

server_info_in_response

Defines whether to include server information in the Server header.

server_info_in_response: bool = False

lifespan

A list of background tasks that start and stop along with the server.

lifespan: Optional[List[LifespanCTX]] = None

on_startup

A list of tasks executed immediately after the application starts.

on_startup: Optional[List[LifespanHook]] = None

on_shutdown

Tasks executed when the server stops.

on_shutdown: Optional[List[LifespanHook]] = None

on_cleanup

Tasks executed after on_shutdown.

on_cleanup: Optional[List[LifespanHook]] = None

http_route_handlers

HTTP routers that can be either individual handlers or groups of HTTPRouter.

http_route_handlers: Iterable[HTTPRouterType] = ()

DI Attributes (dishka)

Rapidy uses the Dishka library as its dependency injection engine.

Learn more about the DI mechanism and integration details here.

di_container

An external DI container that can be passed to Rapidy.

di_container: AsyncContainer | None = None

By default, Rapidy creates and manages its own container. If you pass a container manually, you must manage its lifecycle yourself (startup and shutdown).

Rapidy will not create a new container even if other DI parameters are specified.

Dishka documentation — container.