跳转至

路由 API

routes

路由管理模块

提供了 FastAPI 路由的统一返回格式和自动发现功能

ApiResponse

Bases: BaseModel

统一返回格式

Source code in faster_app/routes/base.py
class ApiResponse(BaseModel):
    """统一返回格式"""

    message: str
    data: dict

RoutesDiscover

Bases: BaseDiscover

Source code in faster_app/routes/discover.py
class RoutesDiscover(BaseDiscover):
    INSTANCE_TYPE = APIRouter
    TARGETS = [
        {
            "directory": "apps",
            "filename": None,
            "skip_dirs": ["__pycache__"],
            "skip_files": [],
        },
        {
            "directory": f"{BASE_DIR}/routes/builtins",
            "filename": None,
            "skip_dirs": ["__pycache__"],
            "skip_files": [],
        },
        # 默认不加载内置路由样例, 开发环境反注释代码进行调试
        # {
        #     "directory": f"{BASE_DIR}/apps",
        #     "filename": None,
        #     "skip_dirs": ["__pycache__"],
        #     "skip_files": [],
        # },
    ]

    def import_and_extract_instances(
        self, file_path: str, module_name: str
    ) -> list[APIRouter]:
        """
        导入模块并提取路由实例
        对于路由, 我们查找已经实例化的 APIRouter 对象
        """
        instances = []

        try:
            # 动态导入模块
            import importlib.util
            import inspect

            spec = importlib.util.spec_from_file_location(module_name, file_path)
            if spec is None or spec.loader is None:
                return instances

            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            # 查找模块中所有的 APIRouter 实例
            for _, obj in inspect.getmembers(module):
                if isinstance(obj, self.INSTANCE_TYPE):
                    instances.append(obj)

        except Exception as e:
            # 静默跳过导入失败的模块, 避免阻断整个发现过程
            print(f"Warning: Failed to import routes from {module_name}: {e}")

        return instances

discover()

自动扫描 TARGETS 中的目录和文件, 导出所有的实例

源代码位于: faster_app/utils/discover.py
def discover(self) -> List[type]:
    """
    自动扫描 TARGETS 中的目录和文件,
    导出所有的实例
    """
    instances = []

    # 扫描 TARGETS 中的目录和文件
    for target in self.TARGETS:
        instances.extend(
            self.scan(
                directory=target.get("directory"),
                filename=target.get("filename"),
                skip_files=target.get("skip_files"),
                skip_dirs=target.get("skip_dirs"),
            )
        )
    return instances

walk(directory, filename=None, skip_files=[], skip_dirs=[])

遍历目录下的所有文件

源代码位于: faster_app/utils/discover.py
def walk(
    self,
    directory: str,
    filename: str = None,
    skip_files: List[str] = [],
    skip_dirs: List[str] = [],
) -> List[str]:
    """
    遍历目录下的所有文件
    """
    results = []
    if not os.path.exists(directory) or not os.path.isdir(directory):
        return results

    for root, dirs, files in os.walk(directory):
        # 过滤掉需要跳过的目录, 直接修改 dirs 列表来影响 os.walk 的遍历
        dirs[:] = [d for d in dirs if d not in skip_dirs]

        for file in files:
            if filename is None or file == filename:
                if file in skip_files:
                    continue
                # 只处理 .py 文件
                if file.endswith(".py"):
                    results.append(os.path.join(root, file))
    return results

scan(directory, filename=None, skip_files=[], skip_dirs=[])

通用扫描方法

参数:

名称 类型 描述 默认
directory str

要扫描的目录路径

必需
filename str

要扫描的具体文件名, 如果为 None 则扫描目录下所有 .py 文件

None
skip_files List[str]

要跳过的文件列表

[]
skip_dirs List[str]

要跳过的目录列表

[]

Returns: 扫描到的所有实例列表

源代码位于: faster_app/utils/discover.py
def scan(
    self,
    directory: str,
    filename: str = None,
    skip_files: List[str] = [],
    skip_dirs: List[str] = [],
) -> List[type]:
    """
    通用扫描方法

    Args:
        directory: 要扫描的目录路径
        filename: 要扫描的具体文件名, 如果为 None 则扫描目录下所有 .py 文件
        skip_files: 要跳过的文件列表
        skip_dirs: 要跳过的目录列表
    Returns:
        扫描到的所有实例列表
    """
    instances = []

    files = self.walk(directory, filename, skip_files, skip_dirs)

    for file in files:
        instances.extend(
            self.import_and_extract_instances(file, file.split("/")[-1][:-3])
        )

    return instances

import_and_extract_instances(file_path, module_name)

导入模块并提取路由实例 对于路由, 我们查找已经实例化的 APIRouter 对象

源代码位于: faster_app/routes/discover.py
def import_and_extract_instances(
    self, file_path: str, module_name: str
) -> list[APIRouter]:
    """
    导入模块并提取路由实例
    对于路由, 我们查找已经实例化的 APIRouter 对象
    """
    instances = []

    try:
        # 动态导入模块
        import importlib.util
        import inspect

        spec = importlib.util.spec_from_file_location(module_name, file_path)
        if spec is None or spec.loader is None:
            return instances

        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # 查找模块中所有的 APIRouter 实例
        for _, obj in inspect.getmembers(module):
            if isinstance(obj, self.INSTANCE_TYPE):
                instances.append(obj)

    except Exception as e:
        # 静默跳过导入失败的模块, 避免阻断整个发现过程
        print(f"Warning: Failed to import routes from {module_name}: {e}")

    return instances

options: show_root_heading: true heading_level: 2