router = APIRouter()

@router.get("/")
def index():
return "/index"

@router.get 背后

訪問起源碼可以看見調(diào)到self.add_api_route傳了一個(gè)func，這個(gè)func在上面的代碼中就是index這個(gè)函數(shù) / endpoint，到這一層就夠了
通過上面得知每個(gè)router.xx后面都調(diào)用了add_api_route，那要實(shí)現(xiàn)類視圖，需要步驟如下：1. 創(chuàng)建APIRouter示例，2. 將類中的某些方法add_api_route加到APIRouter實(shí)例中即可

Controller 裝飾器

這里簡化了參數(shù)描述，會缺少代碼提示，實(shí)際和APIRouter的參數(shù)定義一致就行

class Controller:



    def __init__(self, **kwargs):

        """kwargs 等同于APIRouter 實(shí)例化入?yún)?""

        self.kwargs = kwargs



    def __call__(self, cls):

        # 創(chuàng)建router實(shí)例

        router: APIRouter = APIRouter(

                                      **self.kwargs

                                      )

        # 返回被裝飾類的所有方法和屬性名稱

        for attr_name in dir(cls):

            # 通過反射拿到對應(yīng)屬性的值 或方法對象本身

            attr = getattr(cls, attr_name)

            # 簡單處理，如果函數(shù)返回的是個(gè)字典并且endpoint在這里面，就添加到router上

            if isinstance(attr, dict) and "endpoint" in attr:

                router.add_api_route(**attr)

        # 然后把router 給被裝飾的類上，再返回被裝飾類

        cls.router = router

        return cls

RequestMapping 裝飾器

被裝飾方法返回帶endpoint的字典，讓Controller能夠掃描到

class RequestMapping:

    """請求"""



    def __init__(self, **kwargs):

        self.kwargs = kwargs



    def __call__(self, func):

        # 這里這個(gè)endpoint 對應(yīng)的value 就是被裝飾的函數(shù)

        # 返回的內(nèi)容其實(shí)是符合self.api_add_route的入?yún)⒁?br />
        return {"endpoint": func, **self.kwargs}

測試代碼 main.py

from fastapi import APIRouter



class Controller:



    def __init__(self, **kwargs):

        """kwargs 等同于APIRouter 實(shí)例化入?yún)?""

        self.kwargs = kwargs



    def __call__(self, cls):

        # 創(chuàng)建router實(shí)例

        router: APIRouter = APIRouter(

            **self.kwargs

        )

        # 返回被裝飾類的所有方法和屬性名稱

        for attr_name in dir(cls):

            # 通過反射拿到對應(yīng)屬性的值 或方法對象本身

            attr = getattr(cls, attr_name)

            # 簡單處理，如果函數(shù)返回的是個(gè)字典并且endpoint在這里面，就添加到router上

            if isinstance(attr, dict) and "endpoint" in attr:

                router.add_api_route(**attr)

        # 然后把router 給被裝飾的類上，再返回被裝飾類

        cls.router = router

        return cls



class RequestMapping:

    """請求"""



    def __init__(self, **kwargs):

        self.kwargs = kwargs



    def __call__(self, func):

        # 這里這個(gè)endpoint 對應(yīng)的value 就是被裝飾的函數(shù)

        # 返回的內(nèi)容其實(shí)是符合self.api_add_route的入?yún)⒁?br />
        return {"endpoint": func, **self.kwargs}



def get_db():

    print("模擬db session")

    return "模擬db session"



@Controller(prefix="/demo", tags=["demo"])

class Demo:



    name: str = "ggbond"

    db_session: str = Depends(get_db)



    @RequestMapping(path="")

    def get_list(self):

        return [self.name, self.db_session]



from fastapi import FastAPI



app = FastAPI()



app.include_router(Demo.router)



if __name__ == '__main__':

    import uvicorn



    uvicorn.run(app)

問題

遇到的這些問題，解決方法其實(shí)之前大佬都寫好了 https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py

將self識別成了查詢參數(shù)

解決方法 update_endpoint_signature

將self這個(gè)方法簽名修改成Depends(cls)– 實(shí)例化當(dāng)前類對象，并依賴注入（依賴注入的參數(shù)，無需用戶傳遞哦）。

def update_endpoint_signature(endpoint, cls_obj):

    """

    source：https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py#L92

    修改端點(diǎn)簽名，將self參數(shù)改成關(guān)鍵字參數(shù) self = Depends(cls)

    :param endpoint: 端點(diǎn)

    :param cls_obj: 被裝飾的類 本身

    :return:

    """

    # 獲取原始端點(diǎn)函數(shù)的簽名

    old_signature = inspect.signature(endpoint)

    # 獲取參數(shù)列表

    old_parameters = list(old_signature.parameters.values())

    # 獲取原始端點(diǎn)函數(shù)的第一個(gè)參數(shù)， self

    old_first_parameter = old_parameters[0]

    # 將第一個(gè)參數(shù)替換為依賴注入 當(dāng)前類 的參數(shù)

    # 

    new_first_parameter = old_first_parameter.replace(default=Depends(cls_obj))

    new_parameters = [new_first_parameter] + [

        parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY)

        for parameter in old_parameters[1:]

    ]  # 替換其他參數(shù)的類型為 KEYWORD_ONLY，以便正確執(zhí)行依賴注入

    new_signature = old_signature.replace(parameters=new_parameters)  # 創(chuàng)建新的簽名

    setattr(endpoint, "__signature__", new_signature)  # 替換端點(diǎn)函數(shù)的簽名

依賴注入未被正確解析

解決方法

需要修改類的簽名、和init方法，并在init方法時(shí)調(diào)用依賴注入的函數(shù)這里的實(shí)例get_db，并把結(jié)果更新到db_session中去

CBV_CLASS_KEY = "__cbv_class__"



def _init_cbv(cls: type[Any]) -> None:

    """

    Idempotently modifies the provided cls, performing the following modifications:
    * The __init__ function is updated to set any class-annotated dependencies as instance attributes
    * The __signature__ attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
    """
    # 實(shí)例化之后就不用實(shí)例化了
    if getattr(cls, CBV_CLASS_KEY, False):  # pragma: no cover
        return  # Already initialized
    # 保留原本的 init
    old_init: Callable[..., Any] = cls.__init__
    # 獲取原本的 實(shí)例簽名
    old_signature = inspect.signature(old_init)
    # 刪除self
    old_parameters = list(old_signature.parameters.values())[1:]  # drop self parameter
    # 改成關(guān)鍵字參數(shù)
    new_parameters = [
        x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
    ]
    # 依賴注入列表
    dependency_names: list[str] = []
    # 返回類 cls 中的所有類型標(biāo)注，
    for name, hint in get_type_hints(cls).items():
        if is_classvar(hint):
            continue
        parameter_kwargs = {"default": getattr(cls, name, Ellipsis)}
        dependency_names.append(name)
        # 加入到關(guān)鍵字參數(shù)中
        new_parameters.append(
            inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
        )
    # 替換簽名參數(shù)
    new_signature = old_signature.replace(parameters=new_parameters)

    # 實(shí)例化時(shí)候的處理
    def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
        for dep_name in dependency_names:
            # pop 會拿到依賴注入的返回值
            dep_value = kwargs.pop(dep_name)
            # 設(shè)置成屬性
            setattr(self, dep_name, dep_value)
        # 再調(diào)用原來的實(shí)例化方法
        old_init(self, *args, **kwargs)

    setattr(cls, "__signature__", new_signature)
    setattr(cls, "__init__", new_init)

完整實(shí)現(xiàn)

boot.py

"""

Source: https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py

"""



from __future__ import annotations



import inspect

from collections.abc import Callable

from typing import Any, TypeVar, get_type_hints



from fastapi import APIRouter, Depends

from pydantic.v1.typing import is_classvar



T = TypeVar("T")



CBV_CLASS_KEY = "__cbv_class__"



def _init_cbv(cls: type[Any]) -> None:

    """

    Idempotently modifies the provided cls, performing the following modifications:
    * The __init__ function is updated to set any class-annotated dependencies as instance attributes
    * The __signature__ attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
    """
    if getattr(cls, CBV_CLASS_KEY, False):  # pragma: no cover
        return  # Already initialized
    old_init: Callable[..., Any] = cls.__init__
    old_signature = inspect.signature(old_init)
    old_parameters = list(old_signature.parameters.values())[1:]  # drop self parameter
    new_parameters = [
        x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
    ]
    dependency_names: list[str] = []
    for name, hint in get_type_hints(cls).items():
        if is_classvar(hint):
            continue
        parameter_kwargs = {"default": getattr(cls, name, Ellipsis)}
        dependency_names.append(name)
        new_parameters.append(
            inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
        )
    new_signature = old_signature.replace(parameters=new_parameters)

    def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
        for dep_name in dependency_names:
            dep_value = kwargs.pop(dep_name)
            setattr(self, dep_name, dep_value)
        old_init(self, *args, **kwargs)

    setattr(cls, "__signature__", new_signature)
    setattr(cls, "__init__", new_init)
    setattr(cls, CBV_CLASS_KEY, True)

def _update_cbv_route_endpoint_signature(cls: type[Any], endpoint: Callable[..., Any]) -> None:
    """
    Fixes the endpoint signature for a cbv route to ensure FastAPI performs dependency injection properly.
    """
    old_signature = inspect.signature(endpoint)
    old_parameters: list[inspect.Parameter] = list(old_signature.parameters.values())
    old_first_parameter = old_parameters[0]
    new_first_parameter = old_first_parameter.replace(default=Depends(cls))
    new_parameters = [new_first_parameter] + [
        parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY) for parameter in old_parameters[1:]
    ]
    new_signature = old_signature.replace(parameters=new_parameters)
    setattr(endpoint, "__signature__", new_signature)

class Controller:

    def __init__(self, **kwargs):
        """kwargs 等同于APIRouter 實(shí)例化入?yún)?""
        self.kwargs = kwargs

    def __call__(self, cls):
        _init_cbv(cls)
        # 創(chuàng)建router實(shí)例
        router: APIRouter = APIRouter(
            **self.kwargs
        )
        # 返回被裝飾類的所有方法和屬性名稱
        for attr_name in dir(cls):
            # 通過反射拿到對應(yīng)屬性的值 或方法對象本身
            attr = getattr(cls, attr_name)
            # 添加到router上
            if isinstance(attr, BaseRoute) and hasattr(attr, "kwargs"):
                _update_cbv_route_endpoint_signature(cls, attr.kwargs["endpoint"])

                if isinstance(attr, RequestMapping):
                    router.add_api_route(**attr.kwargs)
                elif isinstance(attr, WebSocket):
                    router.add_websocket_route(**attr.kwargs)
                else:
                    assert False, "Cls Type is RequestMapping or WebSocket"

        cls.router = router
        return cls

class BaseRoute:
    pass

class RequestMapping(BaseRoute):
    """請求"""

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    def __call__(self, func):
        # 這里這個(gè)endpoint 對應(yīng)的value 就是被裝飾的函數(shù)
        # 返回的內(nèi)容其實(shí)是符合self.api_add_route的入?yún)⒁?        self.kwargs["endpoint"] = func
        return self

class WebSocket(BaseRoute):

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    def __call__(self, func):
        # 這里這個(gè)endpoint 對應(yīng)的value 就是被裝飾的函數(shù)
        # 返回的內(nèi)容其實(shí)是符合self.api_add_route的入?yún)⒁?        self.kwargs["endpoint"] = func
        return self

main.py

from fastapi import Depends



from boot import Controller, RequestMapping, WebSocket



def get_db():

    print("模擬db session")

    return "模擬db session"



@Controller(prefix="/demo", tags=["demo"])

class Demo:

    name = "ggbond"

    # 帶類型標(biāo)注類屬性會被加入到方法中如router.get(db_session: str = Depends(get_db))

    db_session: str = Depends(get_db)



    @RequestMapping(path="")

    def get_list(self, age: int):

        return [self.name, self.db_session]



    @WebSocket(path="/ws")

    async def ws(self):

        pass



from fastapi import FastAPI



app = FastAPI()



app.include_router(Demo.router)



if __name__ == '__main__':

    import uvicorn



    uvicorn.run(app)

文章轉(zhuǎn)自微信公眾號@7y記

上一篇：

FastAPI：重燃Python Web開發(fā)的火花

下一篇：

FastAPI，一個(gè)快速開發(fā) API 的 Python 框架！
#你可能也喜歡這些API文章!

我們有何不同？

API服務(wù)商零注冊

多API并行試用

數(shù)據(jù)驅(qū)動選型，提升決策效率

查看全部API→
??

熱門場景實(shí)測，選對API

#AI文本生成大模型API

對比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力

25個(gè)渠道
一鍵對比試用API 限時(shí)免費(fèi)

#AI深度推理大模型API

對比大模型API的邏輯推理準(zhǔn)確性、分析深度、可視化建議合理性

10個(gè)渠道
一鍵對比試用API 限時(shí)免費(fèi)