FastAPI 新手必看:如何搭建一個清晰的項目骨架?
你是否也曾有過這樣的困惑?在沉浸於 FastAPI 的官方文檔和各類教程後,掌握了 CRUD、依賴注入等基礎知識,正摩拳擦掌準備構建自己的第一個小型項目時,卻突然發現:功能雖小,文件卻越堆越多,凌亂得像一團麻線?官方文檔在 "教程 - 用戶指南" 裏的 "更大的應用 - 多個文件" 這個章節裏雖然講到了,但那些抽象的概念,似乎總是無法與實際的代碼結構完全匹配。別擔心,這正是許多 FastAPI 初學者都會遇到的 “成長的煩惱”:如何優雅地組織項目,讓代碼清晰、易懂、便於維護,即便只是一個小型應用?今天,我們就來揭祕 FastAPI 小型項目的最佳實踐,手把手帶你走出結構混亂的迷宮,搭建出既穩固又靈活的骨架!
你可能會覺得,對一個小型項目來說,代碼組織似乎沒那麼重要?畢竟功能簡單,很快就能完成。但事實並非如此!良好的代碼結構就像一幅精確的地圖,能讓你在代碼庫中迅速定位、理解功能;它能大幅提升你的開發效率,減少未來引入新功能或修復 bug 時的阻力;更重要的是,當你的項目需要迭代、擴展甚至開始團隊協作時,清晰的結構能讓整個過程如絲般順滑,避免陷入 “麪條代碼”(spaghetti code)的泥潭。因此,從一開始就建立規範、合理的項目結構,是打造高質量、可維護應用程序的關鍵一步,無論項目大小。
以下是一個小型 FastAPI 項目,帶有 CRUD 和用戶登錄驗證功能的最佳目錄結構示例。這種結構能夠保證代碼清晰、可維護,並適合未來的擴展。(示例代碼使用了 SQLAlchemy 2.0 和 Pydantic v2 基於 SQLite 數據庫。)
項目目錄結構
.
├── app/
│ ├── __init__.py # 初始化模塊
│ ├── main.py # 應用入口文件
│ ├── core/ # 核心配置和工具模塊
│ │ ├── __init__.py
│ │ ├── config.py # 配置文件,如數據庫連接、JWT 密鑰等
│ │ ├── security.py # 安全相關工具(如密碼加密、JWT 生成與驗證)
│ │ └── dependencies.py # 全局依賴項
│ ├── models/ # 數據庫模型
│ │ ├── __init__.py
│ │ ├── base.py # 基礎模型和數據庫連接
│ │ └── user.py # 用戶模型
│ ├── schemas/ # 數據驗證和序列化
│ │ ├── __init__.py
│ │ ├── user.py # 用戶相關 Pydantic 模型
│ ├── crud/ # 數據庫操作(CRUD)
│ │ ├── __init__.py
│ │ └── user.py # 用戶相關 CRUD 操作
│ ├── api/ # API 路由
│ │ ├── __init__.py
│ │ ├── deps.py # API 路由依賴項
│ │ ├── user.py # 用戶相關路由
│ │ └── auth.py # 身份驗證路由
│ ├── tests/ # 測試用例
│ │ ├── __init__.py
│ │ ├── test_auth.py # 身份驗證相關測試
│ │ └── test_user.py # 用戶相關測試
├── .env # 環境變量
├── requirements.txt # 項目依賴
└── alembic/ # 數據庫遷移(可選)
├── env.py
└── versions/
代碼示例
1. main.py
from fastapi import FastAPI
from app.api import auth, user
app = FastAPI()
# 註冊路由
app.include_router(auth.router, prefix="/auth", tags=["Authentication"])
app.include_router(user.router, prefix="/users", tags=["Users"])
2. core/config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
JWT_SECRET: str = "your_jwt_secret"
JWT_ALGORITHM: str = "HS256"
DATABASE_URL: str = "sqlite+aiosqlite:///./test.db"
class Config:
env_file = ".env"
settings = Settings()
3. core/security.py
from passlib.context import CryptContext
from jose import jwt
from datetime import datetime, timedelta
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def create_access_token(data: dict, expires_delta: timedelta = timedelta(minutes=30)):
to_encode = data.copy()
expire = datetime.utcnow() + expires_delta
to_encode.update({"exp": expire})
return jwt.encode(to_encode, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
4. models/base.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from app.core.config import settings
engine = create_async_engine(settings.DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
class Base(DeclarativeBase):
pass
5. models/user.py
from sqlalchemy import String, Integer, Boolean
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
email: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
hashed_password: Mapped[str] = mapped_column(String, nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
6. schemas/user.py
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
email: EmailStr
password: str
class UserResponse(BaseModel):
id: int
email: EmailStr
is_active: bool
model_config = ConfigDict(from_attributes=True)
7. crud/user.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.models.user import User
from app.schemas.user import UserCreate
from app.core.security import hash_password
async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
result = await db.execute(select(User).filter(User.email == email))
return result.scalars().first()
async def create_user(db: AsyncSession, user: UserCreate) -> User:
db_user = User(email=user.email, hashed_password=hash_password(user.password))
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
8. api/auth.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas.user import UserCreate
from app.core.security import verify_password, create_access_token
from app.crud.user import get_user_by_email
from app.core.dependencies import get_db
router = APIRouter()
@router.post("/login")
async def login(email: str, password: str, db: AsyncSession = Depends(get_db)):
user = await get_user_by_email(db, email)
if not user or not verify_password(password, user.hashed_password):
raise HTTPException(status_code=400, detail="Invalid credentials")
token = create_access_token(data={"sub": user.email})
return {"access_token": token, "token_type": "bearer"}
9. core/dependencies.py
from app.models.base import async_session
async def get_db():
async with async_session() as session:
yield session
進階思考:從小到大
本文所介紹的項目結構,對於初學者入門 FastAPI,以及快速構建中小型、MVP(最小可行性產品)項目而言,無疑是一個非常高效且清晰的起點。它能讓你快速組織代碼,實現基本功能,並享受到 FastAPI 帶來的開發便利。
然而,當你的應用逐漸發展壯大,業務邏輯變得愈發複雜,模塊間需要更細粒度的解耦與協作時,你可能會發現當前這種結構會略顯不足。此時,我們便會更推薦採用一種更具擴展性的三層架構——路由層(Router)、服務層(Service)和數據倉庫層(Repository):
-
• 路由層 (Router):負責接收 HTTP 請求,進行初步的參數驗證,並協調調用服務層的方法。它專注於 API 接口的定義和請求 - 響應的處理,保持輕量。
-
• 服務層 (Service):承載核心業務邏輯。它將調用數據倉庫層來執行數據庫操作,並在此基礎上處理複雜的業務規則、數據轉換等,保持與 HTTP 層和數據庫層的解耦。
-
• 數據倉庫層 (Repository):專注於與數據庫的交互,封裝所有的 CRUD(創建、讀取、更新、刪除)操作,對外提供統一的數據訪問接口,使得業務邏輯層無需關心具體的 ORM 細節,例如本示例中的
crud模塊就可以演變爲repository層的一部分。
這種分層模式可以進一步增強代碼的模塊化、可測試性,同時爲未來的擴展和維護提供了更堅實的基礎。希望本文能爲你構建高效、健壯的 FastAPI 應用邁出堅實的第一步,並在未來業務發展時,爲你提供靈活的架構升級思路。
希望本文能爲你構建高效、健壯的 FastAPI 應用提供一份實用的指南。現在,就從一個清晰的項目結構開始,將你的想法變爲現實,讓每一次編碼都更加輕鬆、更有序!
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/GeijxFGFZnFyS6c06Q6GYA