Added AioHttpRouter and examples. Fixed falcon example

Author: mr-fatalyst

CHANGELOG.md

@@ -7,6 +7,7 @@ FastOpenAPI follows the [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ## [0.5.0] - Unreleased
 
 ### Added
+- `AioHttpRouter` for integration with the `AioHttp` framework
 - Class-level cache for model schemas
 
 ## [0.4.0] - 2025-03-20

examples/aiohttp/app/__init__.py

@@ -0,0 +1,6 @@
+from fastopenapi.routers import AioHttpRouter
+
+from .v1 import v1_router
+
+api_router = AioHttpRouter()
+api_router.include_router(v1_router, prefix="/v1")

examples/aiohttp/app/api/__init__.py

@@ -0,0 +1,8 @@
+from fastopenapi.routers import AioHttpRouter
+
+from .authors import router as authors_router
+from .posts import router as posts_router
+
+v1_router = AioHttpRouter()
+v1_router.include_router(authors_router)
+v1_router.include_router(posts_router)

examples/aiohttp/app/api/routes.py

@@ -0,0 +1,48 @@
+from aiohttp import web
+
+from fastopenapi.routers import AioHttpRouter
+
+from ...schemas.authors import (
+    AuthorSchema,
+    CreateAuthorSchema,
+    FilterAuthorSchema,
+    UpdateAuthorSchema,
+)
+from ...services.authors import AuthorService
+
+author_service = AuthorService()
+
+router = AioHttpRouter()
+
+
+@router.post("/authors", tags=["Authors"], status_code=201, response_model=AuthorSchema)
+async def create_author(body: CreateAuthorSchema) -> AuthorSchema:
+    return await author_service.create_author(body)
+
+
+@router.get("/authors/{author_id}", tags=["Authors"], response_model=AuthorSchema)
+async def get_author(author_id: int) -> AuthorSchema:
+    author = await author_service.get_author(author_id)
+    if not author:
+        raise web.HTTPNotFound(reason="Author not found")
+    return author
+
+
+@router.get("/authors/", tags=["Authors"], response_model=list[AuthorSchema])
+async def get_authors(body: FilterAuthorSchema) -> list[AuthorSchema]:
+    return await author_service.get_authors(body)
+
+
+@router.delete("/authors/{author_id}", tags=["Authors"], status_code=204)
+async def delete_author(author_id: int) -> None:
+    author = await author_service.delete_author(author_id)
+    if not author:
+        raise web.HTTPNotFound(reason="Author not found")
+
+
+@router.patch("/authors/{author_id}", tags=["Authors"], response_model=AuthorSchema)
+async def update_author(author_id: int, body: UpdateAuthorSchema) -> AuthorSchema:
+    author = await author_service.update_author(author_id, body)
+    if not author:
+        raise web.HTTPNotFound(reason="Author not found")
+    return author

examples/aiohttp/app/api/v1/__init__.py

@@ -0,0 +1,47 @@
+from aiohttp import web
+
+from fastopenapi.routers import AioHttpRouter
+
+from ...schemas.posts import (
+    CreatePostSchema,
+    FilterPostSchema,
+    PostSchema,
+    UpdatePostSchema,
+)
+from ...services.posts import PostService
+
+post_service = PostService()
+router = AioHttpRouter()
+
+
+@router.post("/posts", tags=["Posts"], status_code=201, response_model=PostSchema)
+async def create_post(body: CreatePostSchema) -> PostSchema:
+    return await post_service.create_post(body)
+
+
+@router.get("/posts/{post_id}", tags=["Posts"], response_model=PostSchema)
+async def get_post(post_id: int) -> PostSchema:
+    post = await post_service.get_post(post_id)
+    if not post:
+        raise web.HTTPNotFound(reason="Author not found")
+    return post
+
+
+@router.get("/posts/", tags=["Posts"], response_model=list[PostSchema])
+async def get_posts(body: FilterPostSchema) -> list[PostSchema]:
+    return await post_service.get_posts(body)
+
+
+@router.delete("/posts/{post_id}", tags=["Posts"], status_code=204)
+async def delete_post(post_id: int) -> None:
+    post = await post_service.delete_post(post_id)
+    if not post:
+        raise web.HTTPNotFound(reason="Post not found")
+
+
+@router.patch("/posts/{post_id}", tags=["Posts"], response_model=PostSchema)
+async def update_post(post_id: int, body: UpdatePostSchema) -> PostSchema:
+    post = await post_service.update_post(post_id, body)
+    if not post:
+        raise web.HTTPNotFound(reason="Post not found")
+    return post

examples/aiohttp/app/api/v1/authors.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel, Field
+
+
+class AuthorSchema(BaseModel):
+    id: int
+    name: str
+    bio: str | None = None
+
+
+class FilterAuthorSchema(BaseModel):
+    id: int = Field(default=None)
+    name: str = Field(default=None)
+
+
+class CreateAuthorSchema(BaseModel):
+    name: str = Field(..., max_length=50)
+    bio: str | None = Field(None, max_length=200)
+
+
+class UpdateAuthorSchema(BaseModel):
+    name: str = Field(default=None, max_length=50)
+    bio: str = Field(default=None, max_length=200)

examples/aiohttp/app/api/v1/posts.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel, Field
+
+
+class PostSchema(BaseModel):
+    id: int
+    title: str
+    content: str
+    author_id: int
+
+
+class FilterPostSchema(BaseModel):
+    id: int = Field(default=None)
+    title: str = Field(default=None)
+
+
+class CreatePostSchema(BaseModel):
+    title: str = Field(..., max_length=100)
+    content: str
+    author_id: int
+
+
+class UpdatePostSchema(BaseModel):
+    title: str = Field(default=None, max_length=100)
+    content: str = Field(default=None)