fastapi

FastAPI中的路由是通过装饰器（Decorator）的方式来定义的

APIRouter允许你在不同的模块中定义路由操作，然后将它们挂载到全局应用程序路由中

from fastapi import APIRouter

router = APIRouter()

@router.get("/get")
async def get():
    return {'api': 'get'}
    
@router.post("/post")
async def post():
    return {'api': 'post'}
    
@router.put("/put")
async def put():
    return {'api': 'put'}
    
@router.patch("/patch")
async def patch():
    return {'api': 'patch'}
    
@router.delete("/delete")
async def delete():
    return {'api': 'delete'}

路由注册

from fastapi import FastAPI
from routers.router_1 import router

app = FastAPI()
app.include_router(router)

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, World"}

• @app.get("/") 是一个装饰器，它告诉 FastAPI 当收到 GET 请求并且 URL 路径为 "/" 时，调用下面定义的 read_root() 函数来处理请求
• 装饰器的原理是在 FastAPI 内部维护一个路由表，将 URL 路径和请求方法与处理函数进行关联。当收到请求时，FastAPI 根据请求的 URL 和方法查找路由表，并调用对应的处理函数来处理请求

class APIRouter(routing.Router):
    def __init__(
        self,
        routes: List[routing.BaseRoute] = None,
        redirect_slashes: bool = True,
        default: ASGIApp = None,
        dependency_overrides_provider: Any = None,
        route_class: Type[APIRoute] = APIRoute,
        default_response_class: Type[Response] = None,
        on_startup: Sequence[Callable] = None,
        on_shutdown: Sequence[Callable] = None,
    ) -> None:
        super().__init__(
            routes=routes,
            redirect_slashes=redirect_slashes,
            default=default,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
        )
        self.dependency_overrides_provider = dependency_overrides_provider
        self.route_class = route_class
        self.default_response_class = default_response_class

    def get(
        self,
        path: str,
        *,
        response_model: Type[Any] = None,
        status_code: int = 200,
        tags: List[str] = None,
        dependencies: Sequence[params.Depends] = None,
        summary: str = None,
        description: str = None,
        response_description: str = "Successful Response",
        responses: Dict[Union[int, str], Dict[str, Any]] = None,
        deprecated: bool = None,
        operation_id: str = None,
        response_model_include: Union[SetIntStr, DictIntStrAny] = None,
        response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
        response_model_by_alias: bool = True,
        response_model_skip_defaults: bool = None,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = None,
        name: str = None,
        callbacks: List[APIRoute] = None,
    ) -> Callable:
        if response_model_skip_defaults is not None:
            warning_response_model_skip_defaults_deprecated()  # pragma: nocover
        return self.api_route(
            path=path,
            response_model=response_model,
            status_code=status_code,
            tags=tags or [],
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses or {},
            deprecated=deprecated,
            methods=["GET"],
            operation_id=operation_id,
            response_model_include=response_model_include,
            response_model_exclude=response_model_exclude,
            response_model_by_alias=response_model_by_alias,
            response_model_exclude_unset=bool(
                response_model_exclude_unset or response_model_skip_defaults
            ),
            response_model_exclude_defaults=response_model_exclude_defaults,
            response_model_exclude_none=response_model_exclude_none,
            include_in_schema=include_in_schema,
            response_class=response_class or self.default_response_class,
            name=name,
            callbacks=callbacks,
        )

def api_route(
        self,
        path: str,
        *,
        response_model: Type[Any] = None,
        status_code: int = 200,
        tags: List[str] = None,
        dependencies: Sequence[params.Depends] = None,
        summary: str = None,
        description: str = None,
        response_description: str = "Successful Response",
        responses: Dict[Union[int, str], Dict[str, Any]] = None,
        deprecated: bool = None,
        methods: List[str] = None,
        operation_id: str = None,
        response_model_include: Union[SetIntStr, DictIntStrAny] = None,
        response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
        response_model_by_alias: bool = True,
        response_model_skip_defaults: bool = None,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = None,
        name: str = None,
        callbacks: List[APIRoute] = None,
    ) -> Callable:
        if response_model_skip_defaults is not None:
            warning_response_model_skip_defaults_deprecated()  # pragma: nocover

        def decorator(func: Callable) -> Callable:
            self.add_api_route(
                path,
                func,
                response_model=response_model,
                status_code=status_code,
                tags=tags or [],
                dependencies=dependencies,
                summary=summary,
                description=description,
                response_description=response_description,
                responses=responses or {},
                deprecated=deprecated,
                methods=methods,
                operation_id=operation_id,
                response_model_include=response_model_include,
                response_model_exclude=response_model_exclude,
                response_model_by_alias=response_model_by_alias,
                response_model_exclude_unset=bool(
                    response_model_exclude_unset or response_model_skip_defaults
                ),
                response_model_exclude_defaults=response_model_exclude_defaults,
                response_model_exclude_none=response_model_exclude_none,
                include_in_schema=include_in_schema,
                response_class=response_class or self.default_response_class,
                name=name,
                callbacks=callbacks,
            )
            return func

        return decorator

    def add_api_route(
        self,
        path: str,
        endpoint: Callable,
        *,
        response_model: Type[Any] = None,
        status_code: int = 200,
        tags: List[str] = None,
        dependencies: Sequence[params.Depends] = None,
        summary: str = None,
        description: str = None,
        response_description: str = "Successful Response",
        responses: Dict[Union[int, str], Dict[str, Any]] = None,
        deprecated: bool = None,
        methods: Optional[Union[Set[str], List[str]]] = None,
        operation_id: str = None,
        response_model_include: Union[SetIntStr, DictIntStrAny] = None,
        response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
        response_model_by_alias: bool = True,
        response_model_skip_defaults: bool = None,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = None,
        name: str = None,
        route_class_override: Optional[Type[APIRoute]] = None,
        callbacks: List[APIRoute] = None,
    ) -> None:
        if response_model_skip_defaults is not None:
            warning_response_model_skip_defaults_deprecated()  # pragma: nocover
        route_class = route_class_override or self.route_class
        route = route_class(
            path,
            endpoint=endpoint,
            response_model=response_model,
            status_code=status_code,
            tags=tags or [],
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses or {},
            deprecated=deprecated,
            methods=methods,
            operation_id=operation_id,
            response_model_include=response_model_include,
            response_model_exclude=response_model_exclude,
            response_model_by_alias=response_model_by_alias,
            response_model_exclude_unset=bool(
                response_model_exclude_unset or response_model_skip_defaults
            ),
            response_model_exclude_defaults=response_model_exclude_defaults,
            response_model_exclude_none=response_model_exclude_none,
            include_in_schema=include_in_schema,
            response_class=response_class or self.default_response_class,
            name=name,
            dependency_overrides_provider=self.dependency_overrides_provider,
            callbacks=callbacks,
        )
        self.routes.append(route)

自定义装饰器添加到FastAPI路由

from fastapi import Request, HTTPException
from fastapi import Request
from fastapi import APIRouter,Depends
from sqlalchemy.orm import Session
from app.api import deps

def custom_decorator(view_func):
    async def wrapper(request: Request):
        # 添加自定义逻辑
        # 例如，验证用户是否有权限访问该路由
        logger.info("用户验证访问路由")
        if not user_has_permission(request):
            logger.info("用户验证访问路由失败")
            raise HTTPException(status_code=403, detail="Forbidden")
        else:
            logger.info("用户验证访问路由通过")
        
        # 执行视图函数
        response = await view_func(request)
        
        # 添加自定义逻辑
        # 例如，记录日志
        # log_request(request, response)
        
        return response
    
    def user_has_permission(request: Request):
        # 实现用户权限验证逻辑
        # 例如，从请求中获取用户信息，并检查其权限
        return True  # 示例中始终返回 True，表示权限通过
    
    return wrapper

from fastapi import Request

router = APIRouter()
@router.get("/test")
@custom_decorator
async def test_fastapi_router(request: Request,db: Session = Depends(deps.get_db)):
    logger.info("测试函数执行结束")

# 执行结果
# 2024-04-26 10:11:26.043 | INFO     | app.api.api_v1.endpoints.platform.shangong.project:wrapper:43 - 用户验证访问路由
# 2024-04-26 10:11:26.043 | INFO     | app.api.api_v1.endpoints.platform.shangong.project:wrapper:48 - 用户验证访问路由通过
# 2024-04-26 10:11:26.043 | INFO     | app.api.api_v1.endpoints.platform.shangong.project:test_fastapi_router:72 - 测试函数执行结束