feat: add class-based view and route support, consistent with official FastAPI style

fastapi/cbx.py

@@ -0,0 +1,120 @@
+import inspect
+import logging
+from collections.abc import Callable
+from functools import partial
+from typing import Any
+
+from fastapi import APIRouter
+from typing_extensions import Self
+
+
+class CBV:
+    def __init__(
+        self,
+        cls: type[Any],
+        router: APIRouter,
+        prefix: str = "",
+    ):
+        self.logger = logging.getLogger(self.__class__.__name__)
+        self.router = router
+        self.prefix = prefix
+        self.cls = cls
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Self:
+        self.instance = self.cls(*args, **kwargs)
+        for name in [
+            "head",
+            "get",
+            "post",
+            "put",
+            "delete",
+            "patch",
+            "options",
+            "trace",
+            "connect",
+        ]:
+            if hasattr(self.instance, name):
+                method = getattr(self.instance, name)
+                self.router.add_api_route(
+                    path=self.prefix,
+                    endpoint=method,
+                    methods=[name.upper()],
+                )
+
+        return self
+
+
+class CBR:
+    def __init__(
+        self,
+        cls: type[Any],
+        router: APIRouter,
+        prefix: str = "",
+    ):
+        self.logger = logging.getLogger(self.__class__.__name__)
+        self.router = router
+        self.prefix = prefix
+        self.cls = cls
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Self:
+        self.instance = self.cls(*args, **kwargs)
+
+        for _name, endpoint in inspect.getmembers(
+            self.instance,
+            lambda x: inspect.ismethod(x) or inspect.isfunction(x)
+        ):
+            if cbx_router := endpoint.__annotations__.get("cbx_router"):
+                self.router.add_api_route(
+                    path=f"{self.prefix}{cbx_router['path']}",
+                    endpoint=endpoint,
+                    methods=[cbx_router["method"]],
+                )
+        return self
+
+
+class cbv:
+    def __init__(self, router: APIRouter, prefix: str = ""):
+        self.router = router
+        self.prefix = prefix
+
+    def __call__(self, cls: type[Any]) -> CBV:
+        return CBV(cls, self.router, self.prefix)
+
+
+class cbr:
+
+    class method:
+        def __init__(self, method: str, path: str, *args: Any, **kwargs: Any):
+            self.method = method
+            self.path = path
+            self.args = args
+            self.kwargs = kwargs
+
+        def __call__(self, endpoint: Callable[..., Any]) -> Callable[..., Any]:
+            endpoint.__annotations__.setdefault(
+                "cbx_router",
+                {
+                    "method": self.method,
+                    "path": self.path,
+                    "args": self.args,
+                    "kwargs": self.kwargs,
+                },
+            )
+            return endpoint
+
+    head = partial(method, "HEAD")
+    get = partial(method, "GET")
+    post = partial(method, "POST")
+    put = partial(method, "PUT")
+    delete = partial(method, "DELETE")
+    patch = partial(method, "PATCH")
+    options = partial(method, "OPTIONS")
+    trace = partial(method, "TRACE")
+    connect = partial(method, "CONNECT")
+
+    def __init__(self, router: APIRouter, prefix: str = ""):
+        self.router = router
+        self.prefix = prefix
+
+    def __call__(self, cls: type[Any]) -> CBR:
+        return CBR(cls, self.router, self.prefix)

tests/test_cbx.py

@@ -0,0 +1,95 @@
+from fastapi import APIRouter, FastAPI
+from fastapi.cbx import cbr, cbv
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+
+# ==============================
+# 请求体模型（用于 POST 传参）
+# ==============================
+class UserCreate(BaseModel):
+    username: str
+    age: int
+
+
+# ==============================
+# 1. CBV 测试：自动路由 + 接口传参
+# ==============================
+@cbv(prefix="/cbv", router=APIRouter())
+class CBVTest:
+    def __init__(self, message: str):
+        self.message = message
+
+    # GET 传参：查询参数
+    def get(self, user_id: int, name: str = None):
+        return {"hello": self.message, "user_id": user_id, "name": name}
+
+    # POST 传参：请求体
+    def post(self, user: UserCreate):
+        return {
+            "method": "POST",
+            "message": self.message,
+            "user": user.model_dump(),  # 🔥 修复在这里
+        }
+
+
+# ==============================
+# 2. CBR 测试：装饰器路由 + 接口传参
+# ==============================
+@cbr(prefix="/cbr", router=APIRouter())
+class CBRTest:
+    def __init__(self, service_name: str):
+        self.service_name = service_name
+
+    @cbr.get("/info")
+    def get_info(self, id: int):
+        return {"service": self.service_name, "id": id}
+
+    @cbr.post("/create")
+    def create_user(self, user: UserCreate):
+        return {
+            "service": self.service_name,
+            "user": user.model_dump(),  # 🔥 修复在这里
+        }
+
+
+# ==============================
+# ✅ 优雅初始化（传参）+ 挂载路由
+# ==============================
+cbv_instance = CBVTest(message="cbv_success")
+cbr_instance = CBRTest(service_name="cbr_service")
+
+app = FastAPI()
+app.include_router(cbv_instance.router)
+app.include_router(cbr_instance.router)
+client = TestClient(app)
+
+
+# ==============================
+# 🟢 CBV 测试用例（带真实传参）
+# ==============================
+def test_cbv_get_with_params():
+    resp = client.get("/cbv?user_id=100&name=test")
+    assert resp.status_code == 200
+    assert resp.json()["hello"] == "cbv_success"
+
+
+def test_cbv_post_with_body():
+    resp = client.post("/cbv", json={"username": "test", "age": 20})
+    assert resp.status_code == 200
+    assert resp.json()["user"]["username"] == "test"
+
+
+# ==============================
+# 🟢 CBR 测试用例（带真实传参）
+# ==============================
+def test_cbr_get_with_query():
+    resp = client.get("/cbr/info?id=999")
+    assert resp.status_code == 200
+    assert resp.json()["service"] == "cbr_service"
+
+
+def test_cbr_post_with_body():
+    resp = client.post("/cbr/create", json={"username": "cbr_user", "age": 25})
+    assert resp.status_code == 200
+    assert resp.json()["user"]["age"] == 25