from kaya import KayaApp, HttpContext, HttpMethod, WebSocket from typing import List class Hello(KayaApp): pass # async def handle_request(self, ctx: HttpContext) -> None: # async for chunk in ctx.request_body: # print(chunk.decode()) # await ctx.send_str(200, 'Hello World') app = KayaApp() @app.GET('/hello') @app.GET('/hello2') async def handle_request(ctx: HttpContext) -> None: async for chunk in ctx.request_body: print(chunk.decode()) await ctx.send_str(200, 'Hello World') @app.GET('/hello/alice') async def hello_alice(ctx: HttpContext) -> None: async for _ in ctx.request_body: pass await ctx.send_str(200, 'Hello Alice 2') @app.GET('/hello/*') async def hello_wildcard(ctx: HttpContext, args: List[str]) -> None: async for _ in ctx.request_body: pass await ctx.send_str(200, f'Hello {args[0]}') @app.GET('/hello/bob/marley') async def hello_bob_marley(ctx: HttpContext) -> None: async for _ in ctx.request_body: pass await ctx.send_str(200, f'Hello bob Marley') @app.route('/hello/*', HttpMethod.PUT, recursive=True) async def handle_put_request(ctx: HttpContext, _: List[str]) -> None: async for chunk in ctx.request_body: print(chunk.decode()) await ctx.send_str(200, 'Message received') @app.websocket('/echo') async def echo(ws: WebSocket) -> None: await ws.accept() async for msg in ws: if msg.kind == 'text': await ws.send_text(f"echo: {msg.data}") elif msg.kind == 'binary': data = msg.data assert isinstance(data, bytes) await ws.send_bytes(data)