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.
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])
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)
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.
lifespan
A list of background tasks that start and stop along with the server.
on_startup
A list of tasks executed immediately after the application starts.
on_shutdown
Tasks executed when the server stops.
on_cleanup
Tasks executed after on_shutdown.
http_route_handlers
HTTP routers that can be either individual handlers or groups of HTTPRouter.
aiohttp Attributes
middlewares
A list of middleware applied to all handlers, including child applications.
client_max_size
Maximum request size in bytes.
logger
Logger for receiving logs from Application.
Running the Application
Simple Run
Copy the following code into 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 Run (Gunicorn)
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)
The gunicorn main:app command refers to:
main: themain.pyfile (Python module).rapidy: the object created insidemain.pyin the linerapidy = Rapidy().--reload: restarts the server when the code changes. Use only for development.