Logo

菜单

发布于 2026-01-06 2 阅读
0

如何构建大型 FastAPI 项目

如何构建大型 FastAPI 项目

整套配置方案都以模板的形式发布在GitHub上。

我刚开始使用 FastAPI 时，发现要合理地组织项目结构非常困难。对于大型项目来说，把所有内容都写在一两个文件中并不是理想的解决方案。

因此，基本要求是创建一个易于理解的项目结构，并能很好地集成 SQLAlchemy 和单元测试。此外，我还希望通过使用不同的模块来实现清晰的关注点分离。

基本文件夹结构

我们将项目结构分为两个模块：src 和 test：

.
└── backend/
    ├── src/
    │   ├── config/
    │   ├── routes/
    │   ├── util/
    │   └── __init__.py
    ├── test/
    │   ├── routes/
    │   ├── test_util/
    │   ├── __init.py__
    │   ├── app_test.py
    │   └── conftest.py
    ├── main.py
    └── requirements.txt
Enter fullscreen mode Exit fullscreen mode

基本 FastAPI

应用程序启动文件是main.py. 在此文件中，我们可以按预期定义 FastAPI 服务器：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import RedirectResponse

app = FastAPI(
    title=APP_NAME,
    version=VERSION
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True
)

# Redirect / -> Swagger-UI documentation
@app.get("/")
def main_function():
    """
    # Redirect
    to documentation (`/docs/`).
    """
    return RedirectResponse(url="/docs/")
Enter fullscreen mode Exit fullscreen mode

一些通用应用程序参数定义在.config.py文件中/config。这些值也可以在部署过程中作为环境变量。

APP_NAME = "My fancy app"
VERSION = "v1.0.0"
Enter fullscreen mode Exit fullscreen mode

路线

在路由方面，我从 SvelteKit 中汲取了灵感。他们使用基于文件的路由，这使得查找特定代码段变得非常简单——你只需要按照 URL 的结构遍历文件即可。这对于 API 来说似乎也是一个不错的想法，所以我采用了这种方法，尽管它的缺点是会反复创建同名文件。

假设我们构建一个非常基础的会议室预订 API。我们可能会遇到以下几个接口：

/rooms          GET
                POST
                DELETE
                PUT

/bookings       GET
                POST
                DELETE
                PUT

/users          GET
                POST
                DELETE
                PUT
Enter fullscreen mode Exit fullscreen mode

以及以下数据库表：

rooms           bookings                users
=========       =============           =====
id              id                      id
name            booker_id               name
capacity        starting_time           mail
building        ending_time
                room_id
Enter fullscreen mode Exit fullscreen mode

在这个例子中，我们将在routes-directory 目录下创建三个文件夹：

.
└── backend/
    └── src/
        └── routes/
            ├── users
            ├── bookings
            ├── rooms
            └── __init.py__
Enter fullscreen mode Exit fullscreen mode

此外，我喜欢将身份验证逻辑放在单独的路由中。

每个文件夹内都包含三个文件：

  • main.py：实际 API 端点函数所在的位置，可能如下所示：

    @router.get("/")
async def get_all_users(user: User = 
Depends(get_current_active_user),
                        db: Session = Depends(get_db)):
    """
    # Get a list of all users

    **Access:**
    - Admins get a list of all users.
    - Users with lower rights get a list with only the enabled users.
    """
    if user.super_admin:
        return get_users_admin(db=db)
    else:
        return get_users(db=db)

  • controller.py处理你的逻辑

    def get_user_by_id(user_id: int, db: Session):
    user = db.query(User).filter(User.id == user_id).first()

    if not user:
        raise HTTPException(
            status_code=404,
            detail="There is no user with this id."
        )

    return user

  • schemas.py：与此路由相关的 Pydantic 模式的所在地

要使用这些路由，我们必须为相应的每条路由创建一个路由器{route}/main.py：

router = APIRouter(
    prefix="/users",
    tags=["Users"],
    responses={404: {"description": "Not found"}},
)
Enter fullscreen mode Exit fullscreen mode

然后我们将这些路由器导入到应用程序中main.py：

from src.routes import bookings, users

# ---- Do this for all of your routes ----
app.include_router(users.main.router)
app.include_router(bookings.main.router)
# ----------------------------------------
Enter fullscreen mode Exit fullscreen mode

数据库集成

完成以上步骤后，我们已成功将 FastAPI 服务器组织成多个文件夹。但是，目前仍无法访问数据库。我们将使用 SQLAlchemy 作为 ORM。

根据 FastAPI 文档，我们首先需要在内部指定连接参数src/config/database.py：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

db_username = "user"
db_password = "password123"
db_url = "127.0.0.1:3306"
db_name = "example"

connectionString = f'mariadb+pymysql://{db_username}:{db_password}@{db_url}/{db_name}'

# use echo=True for debugging
engine = create_engine(connectionString, echo=False)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
Enter fullscreen mode Exit fullscreen mode

根据您使用的数据库，您可能需要修改连接字符串并安装不同的 pip 包。对于 MariaDB，相关的 pip 包是mariadb和。pymysqlSQLAlchemy==1.4.36

请注意，SQLAlchemy >= v2.0 引入了重大变更，因此如果您想使用新版本，则需要修改模板！

为了通过名为 `<endpoint_name>` 的端点与数据库进行交互，我们将使用依赖注入。我们在以下位置定义依赖项src/util/db_dependency.py：

from src.config.database import SessionLocal


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
Enter fullscreen mode Exit fullscreen mode

现在我们可以为每个路由创建数据库模型文件，例如src/routes/users/models.py：

from src.config.database import Base
from sqlalchemy import Column, String, Integer


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(length=50))
    mail = Column(String(length=100))
Enter fullscreen mode Exit fullscreen mode

最后，我们将所有模型导入到全局模型中main.py，并从中创建指定的表：

from src.config.database import engine

from src.routes.users import main, models
from src.routes.auth import main, models

users.models.Base.metadata.create_all(bind=engine)
auth.models.Base.metadata.create_all(bind=engine)
Enter fullscreen mode Exit fullscreen mode

单元测试

单元测试的/test文件夹结构与之前相同/src。测试的组织结构也与之前相同。

本文解释了项目结构背后的基本思路。您可以在GitHub上找到完整的模板。该模板还包括身份验证、电子邮件模板（例如，用于通知用户安全问题或发送密码重置链接）以及完整的测试配置。

虽然这种结构在某些项目中对我来说效果不错，但可能还有更好的组织方式。所以，请在评论区告诉我：您对这种项目结构有何看法？您又是如何处理 FastAPI 项目的？

文章来源：https://dev.to/timo_reusch/how-i-struct-big-fastapi-projects-260e
如何使用 JavaScript 和...
创建一个骨架加载