Add kaya-openapi package for automatic OpenAPI spec generation

- New packages/kaya-openapi with OpenAPIMixin, @operation decorator, and generate_spec() that walks the routing tree - Enables kaya-core's Tree.register to expose the original handler callback as an instance attribute for metadata introspection - Registers GET /openapi.json and GET /docs (Swagger UI) routes - Supports and path parameters, docstring descriptions, @operation metadata, and excludes wildcard/WS routes - Adds example/openapi.py, updates CI, README, and requirements
2026-07-25 09:20:19 +00:00
parent 1e1e031a56
commit 338c6bd600
14 changed files with 762 additions and 2 deletions
@@ -0,0 +1,167 @@
+import json
+import unittest
+from typing import Any, Mapping, Sequence
+
+import httpx
+from pwo import async_test
+
+from kaya.core import HttpContext, HttpMethod, KayaApp, WebSocket
+from kaya.openapi import OpenAPIMixin, operation
+
+
+class OpenAPITest(unittest.TestCase):
+    app: KayaApp
+
+    def setUp(self) -> None:
+        self.app = KayaApp(mixins=[OpenAPIMixin(title='Test API', version='1.2.3')])
+
+        @self.app.GET('/hello')
+        async def hello(ctx: HttpContext) -> None:
+            """Say hello."""
+            await ctx.send_str(200, 'Hello World!')
+
+        @self.app.GET('/users/${user_id:int}')
+        @operation(summary='Get a user',
+                   tags=['users'],
+                   responses={
+                       200: {'description': 'The user'},
+                       404: {'description': 'User not found'},
+                   })
+        async def get_user(ctx: HttpContext, user_id: int) -> None:
+            await ctx.send_str(200, str(user_id))
+
+        @self.app.POST('/users/${name}')
+        async def create_user(ctx: HttpContext, name: str) -> None:
+            await ctx.send_str(201, name)
+
+        @self.app.route('/ping')
+        async def ping(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'pong')
+
+        @self.app.GET('/files/*', recursive=True)
+        async def serve_file(ctx: HttpContext, path: Sequence[str]) -> None:
+            await ctx.send_str(200, '/'.join(path))
+
+        @self.app.GET('/internal/health')
+        @operation(hidden=True)
+        async def health(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'ok')
+
+        @self.app.websocket('/echo')
+        async def echo(ws: WebSocket) -> None:
+            await ws.accept()
+            await ws.close()
+
+    async def _get_spec(self) -> Mapping[str, Any]:
+        transport = httpx.ASGITransport(app=self.app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            response = await client.get('/openapi.json')
+            self.assertEqual(200, response.status_code)
+            self.assertEqual('application/json', response.headers['Content-Type'])
+            return json.loads(response.text)
+
+    @async_test
+    async def test_spec_endpoint(self) -> None:
+        spec = await self._get_spec()
+        self.assertEqual('3.1.0', spec['openapi'])
+        self.assertEqual({'title': 'Test API', 'version': '1.2.3'}, spec['info'])
+        self.assertIn('paths', spec)
+
+    @async_test
+    async def test_static_route_with_docstring(self) -> None:
+        spec = await self._get_spec()
+        hello = spec['paths']['/hello']['get']
+        self.assertEqual('Say hello.', hello['description'])
+        self.assertIn('responses', hello)
+
+    @async_test
+    async def test_int_path_parameter(self) -> None:
+        spec = await self._get_spec()
+        operation = spec['paths']['/users/{user_id}']['get']
+        self.assertEqual('Get a user', operation['summary'])
+        self.assertEqual(['users'], operation['tags'])
+        self.assertEqual({
+            '200': {'description': 'The user'},
+            '404': {'description': 'User not found'},
+        }, operation['responses'])
+        self.assertEqual(
+            [{'name': 'user_id', 'in': 'path', 'required': True, 'schema': {'type': 'integer'}}],
+            operation['parameters'])
+
+    @async_test
+    async def test_str_path_parameter(self) -> None:
+        spec = await self._get_spec()
+        operation = spec['paths']['/users/{name}']['post']
+        self.assertEqual(
+            [{'name': 'name', 'in': 'path', 'required': True, 'schema': {'type': 'string'}}],
+            operation['parameters'])
+
+    @async_test
+    async def test_method_agnostic_route(self) -> None:
+        spec = await self._get_spec()
+        path_item = spec['paths']['/ping']
+        for method in ('get', 'put', 'post', 'delete', 'options', 'head', 'patch'):
+            self.assertIn(method, path_item)
+
+    @async_test
+    async def test_excluded_routes(self) -> None:
+        spec = await self._get_spec()
+        paths = spec['paths']
+        # wildcard routes cannot be expressed in OpenAPI
+        self.assertNotIn('/files/*', paths)
+        self.assertFalse(any('files' in path for path in paths))
+        # websocket routes are not part of OpenAPI
+        self.assertNotIn('/echo', paths)
+        # hidden operations are skipped
+        self.assertNotIn('/internal/health', paths)
+        # the mixin's own endpoints are excluded by default
+        self.assertNotIn('/openapi.json', paths)
+        self.assertNotIn('/docs', paths)
+
+    @async_test
+    async def test_docs_endpoint(self) -> None:
+        transport = httpx.ASGITransport(app=self.app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            response = await client.get('/docs')
+            self.assertEqual(200, response.status_code)
+            self.assertEqual('text/html; charset=utf-8', response.headers['Content-Type'])
+            self.assertIn('swagger-ui', response.text)
+            self.assertIn('/openapi.json', response.text)
+
+    @async_test
+    async def test_late_registered_routes_are_included(self) -> None:
+        @self.app.GET('/late')
+        async def late(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'late')
+
+        spec = await self._get_spec()
+        self.assertIn('/late', spec['paths'])
+
+    @async_test
+    async def test_custom_paths_and_self_inclusion(self) -> None:
+        app = KayaApp(mixins=[OpenAPIMixin(title='Custom',
+                                           version='0.1.0',
+                                           spec_path='/spec.json',
+                                           docs_path='/swagger',
+                                           include_self=True)])
+
+        @app.route('/items/${item_id:int}', HttpMethod.DELETE)
+        async def delete_item(ctx: HttpContext, item_id: int) -> None:
+            await ctx.send_empty(204)
+
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            response = await client.get('/spec.json')
+            self.assertEqual(200, response.status_code)
+            spec = json.loads(response.text)
+            self.assertIn('delete', spec['paths']['/items/{item_id}'])
+            self.assertIn('/spec.json', spec['paths'])
+            self.assertIn('/swagger', spec['paths'])
+
+            response = await client.get('/swagger')
+            self.assertEqual(200, response.status_code)
+            self.assertIn('/spec.json', response.text)
+
+
+if __name__ == '__main__':
+    unittest.main()