目錄
01-7. 專案結構最佳實踐
⏱️ 閱讀時間： 15 分鐘 🎯 難度： ⭐⭐ (基礎)
🤔 一句話解釋
好的專案結構讓程式碼易於維護、測試和擴展，是成為資深工程師的必備技能。
📁 從單檔案到多檔案
小型專案（學習/原型）
一個 main.py 就夠了：
my-project/
├── main.py
├── requirements.txt
└── .env中型專案（一般應用）
my-project/
├── app/
│ ├── __init__.py
│ ├── main.py # 應用程式入口
│ ├── config.py # 設定管理
│ ├── database.py # 資料庫連線
│ ├── models.py # 資料庫模型
│ ├── schemas.py # Pydantic 模型
│ ├── crud.py # CRUD 操作
│ └── routers/
│ ├── __init__.py
│ ├── users.py
│ └── items.py
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ └── test_users.py
├── requirements.txt
├── .env
└── .gitignore大型專案（生產環境）
my-project/
├── app/
│ ├── __init__.py
│ ├── main.py # 應用程式入口
│ ├── config.py # 設定管理
│ │
│ ├── api/ # API 路由
│ │ ├── __init__.py
│ │ ├── deps.py # 共用依賴
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── router.py # 路由聚合
│ │ ├── users.py
│ │ ├── items.py
│ │ └── auth.py
│ │
│ ├── core/ # 核心模組
│ │ ├── __init__.py
│ │ ├── security.py # 安全相關
│ │ ├── exceptions.py # 自訂例外
│ │ └── events.py # 啟動/關閉事件
│ │
│ ├── db/ # 資料庫
│ │ ├── __init__.py
│ │ ├── database.py # 連線管理
│ │ └── migrations/ # Alembic 遷移
│ │
│ ├── models/ # 資料庫模型 (SQLAlchemy)
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── user.py
│ │ └── item.py
│ │
│ ├── schemas/ # Pydantic 模型
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ │
│ ├── crud/ # CRUD 操作
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── user.py
│ │ └── item.py
│ │
│ └── services/ # 業務邏輯
│ ├── __init__.py
│ ├── user.py
│ └── email.py
│
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── unit/
│ └── integration/
│
├── alembic/ # 資料庫遷移
├── scripts/ # 工具腳本
├── docker/
│ ├── Dockerfile
│ └── docker-compose.yml
│
├── .env
├── .env.example
├── .gitignore
├── alembic.ini
├── pyproject.toml # 或 requirements.txt
└── README.md🔧 各檔案詳解
app/main.py - 應用程式入口
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.api.v1.router import api_router
from app.core.events import create_start_app_handler, create_stop_app_handler
@asynccontextmanager
async def lifespan(app: FastAPI):
"""應用程式生命週期管理"""
# 啟動時執行
await create_start_app_handler()
yield
# 關閉時執行
await create_stop_app_handler()
def create_application() -> FastAPI:
"""建立 FastAPI 應用程式"""
application = FastAPI(
title=settings.APP_NAME,
description="API 說明",
version=settings.VERSION,
docs_url="/docs" if settings.DEBUG else None,
redoc_url="/redoc" if settings.DEBUG else None,
lifespan=lifespan,
)
# CORS 設定
application.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 註冊路由
application.include_router(api_router, prefix="/api/v1")
return application
app = create_application()app/config.py - 設定管理
from pydantic_settings import BaseSettings
from functools import lru_cache
from typing import List
class Settings(BaseSettings):
"""應用程式設定"""
# 基本設定
APP_NAME: str = "My API"
VERSION: str = "1.0.0"
DEBUG: bool = False
# 伺服器
HOST: str = "0.0.0.0"
PORT: int = 8000
# 資料庫
DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost/db"
# Redis
REDIS_URL: str = "redis://localhost:6379"
# JWT
SECRET_KEY: str = "your-secret-key"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# CORS
ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
case_sensitive = True
@lru_cache()
def get_settings() -> Settings:
"""獲取設定（快取）"""
return Settings()
settings = get_settings()app/api/deps.py - 共用依賴
from typing import AsyncGenerator, Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.database import async_session_maker
from app.core.security import verify_token
from app.models.user import User
from app.crud.user import user_crud
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_db() -> AsyncGenerator[AsyncSession, None]:
"""資料庫 Session 依賴"""
async with async_session_maker() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
async def get_current_user(
db: Annotated[AsyncSession, Depends(get_db)],
token: Annotated[str, Depends(oauth2_scheme)]
) -> User:
"""獲取當前使用者"""
user_id = verify_token(token)
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="無效的認證憑證"
)
user = await user_crud.get(db, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="使用者不存在"
)
return user
# 類型別名
DbSession = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]app/api/v1/router.py - 路由聚合
from fastapi import APIRouter
from app.api.v1 import auth, users, items
api_router = APIRouter()
api_router.include_router(
auth.router,
prefix="/auth",
tags=["認證"]
)
api_router.include_router(
users.router,
prefix="/users",
tags=["使用者"]
)
api_router.include_router(
items.router,
prefix="/items",
tags=["商品"]
)app/api/v1/users.py - 使用者路由
from fastapi import APIRouter, HTTPException, status
from app.api.deps import DbSession, CurrentUser
from app.schemas.user import UserCreate, UserResponse, UserUpdate
from app.crud.user import user_crud
router = APIRouter()
@router.get("/", response_model=list[UserResponse])
async def list_users(
db: DbSession,
skip: int = 0,
limit: int = 100
):
"""列出使用者"""
users = await user_crud.get_multi(db, skip=skip, limit=limit)
return users
@router.get("/me", response_model=UserResponse)
async def get_current_user_info(current_user: CurrentUser):
"""獲取當前使用者資訊"""
return current_user
@router.get("/{user_id}", response_model=UserResponse)
async def get_user(db: DbSession, user_id: int):
"""獲取使用者"""
user = await user_crud.get(db, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="使用者不存在"
)
return user
@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(db: DbSession, user_in: UserCreate):
"""建立使用者"""
existing = await user_crud.get_by_email(db, user_in.email)
if existing:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email 已被使用"
)
user = await user_crud.create(db, obj_in=user_in)
return user
@router.put("/{user_id}", response_model=UserResponse)
async def update_user(
db: DbSession,
user_id: int,
user_in: UserUpdate,
current_user: CurrentUser
):
"""更新使用者"""
if user_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="只能更新自己的資料"
)
user = await user_crud.update(db, db_obj=current_user, obj_in=user_in)
return userapp/schemas/user.py - Pydantic 模型
from pydantic import BaseModel, EmailStr, Field, ConfigDict
from typing import Optional
from datetime import datetime
# ===== 基礎 =====
class UserBase(BaseModel):
"""使用者基礎欄位"""
email: EmailStr
username: str = Field(..., min_length=3, max_length=50)
full_name: Optional[str] = None
# ===== 建立 =====
class UserCreate(UserBase):
"""建立使用者"""
password: str = Field(..., min_length=8)
# ===== 更新 =====
class UserUpdate(BaseModel):
"""更新使用者"""
email: Optional[EmailStr] = None
username: Optional[str] = Field(None, min_length=3, max_length=50)
full_name: Optional[str] = None
password: Optional[str] = Field(None, min_length=8)
# ===== 回應 =====
class UserResponse(UserBase):
"""使用者回應"""
id: int
is_active: bool
created_at: datetime
model_config = ConfigDict(from_attributes=True)
# ===== 資料庫內部 =====
class UserInDB(UserBase):
"""資料庫中的使用者（包含密碼雜湊）"""
id: int
hashed_password: str
model_config = ConfigDict(from_attributes=True)app/crud/base.py - 通用 CRUD
from typing import TypeVar, Generic, Type, Optional, Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from app.db.database import Base
ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
"""通用 CRUD 操作"""
def __init__(self, model: Type[ModelType]):
self.model = model
async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
"""根據 ID 獲取"""
result = await db.execute(
select(self.model).where(self.model.id == id)
)
return result.scalar_one_or_none()
async def get_multi(
self,
db: AsyncSession,
*,
skip: int = 0,
limit: int = 100
) -> Sequence[ModelType]:
"""獲取多筆"""
result = await db.execute(
select(self.model).offset(skip).limit(limit)
)
return result.scalars().all()
async def create(
self,
db: AsyncSession,
*,
obj_in: CreateSchemaType
) -> ModelType:
"""建立"""
db_obj = self.model(**obj_in.model_dump())
db.add(db_obj)
await db.flush()
await db.refresh(db_obj)
return db_obj
async def update(
self,
db: AsyncSession,
*,
db_obj: ModelType,
obj_in: UpdateSchemaType
) -> ModelType:
"""更新"""
update_data = obj_in.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_obj, field, value)
db.add(db_obj)
await db.flush()
await db.refresh(db_obj)
return db_obj
async def delete(self, db: AsyncSession, *, id: int) -> None:
"""刪除"""
obj = await self.get(db, id)
if obj:
await db.delete(obj)
await db.flush()🎯 設計原則
1. 分層架構
┌─────────────────────────────────────┐
│ API Layer (路由) │ ← 處理 HTTP 請求/回應
├─────────────────────────────────────┤
│ Service Layer (服務) │ ← 業務邏輯
├─────────────────────────────────────┤
│ CRUD Layer (CRUD) │ ← 資料存取
├─────────────────────────────────────┤
│ Database Layer (DB) │ ← 資料庫連線
└─────────────────────────────────────┘2. 依賴方向
schemas ← api → crud → models
↓
services
↓
core- schemas: 不依賴其他模組
- models: 不依賴其他模組
- crud: 依賴 models, schemas
- services: 依賴 crud, schemas
- api: 依賴 services, schemas, crud
3. 命名規範
# 檔案命名：小寫 + 底線
user_service.py
auth_router.py
# 類別命名：PascalCase
class UserCreate(BaseModel): ...
class UserService: ...
# 函數命名：小寫 + 底線
async def get_user(): ...
async def create_user(): ...
# 常數命名：大寫 + 底線
DATABASE_URL = "..."
MAX_CONNECTIONS = 10📦 requirements.txt 範例
# Web 框架
fastapi==0.109.0
uvicorn[standard]==0.27.0
# 資料驗證
pydantic==2.5.3
pydantic-settings==2.1.0
email-validator==2.1.0
# 資料庫
sqlalchemy==2.0.25
alembic==1.13.1
asyncpg==0.29.0
# 認證
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
# Redis
redis==5.0.1
# HTTP 客戶端
httpx==0.26.0
# 測試
pytest==7.4.4
pytest-asyncio==0.23.3
pytest-cov==4.1.0
# 開發工具
python-dotenv==1.0.0
black==24.1.1
ruff==0.1.14
mypy==1.8.0✅ 重點總結
專案結構選擇
| 專案規模 | 建議結構 |
|---|---|
| 學習/原型 | 單一 main.py |
| 一般應用 | 基本分層 |
| 生產環境 | 完整分層 + 測試 |
核心原則
- 關注點分離 - 每個模組做一件事
- 依賴注入 - 使用 FastAPI 的 Depends
- 統一命名 - 遵循 Python 命名規範
- 易於測試 - 業務邏輯與框架分離
檔案職責
| 檔案/目錄 | 職責 |
|---|---|
main.py | 應用程式建立和設定 |
config.py | 環境變數和設定 |
api/ | HTTP 路由處理 |
schemas/ | 請求/回應模型 |
models/ | 資料庫模型 |
crud/ | 資料存取層 |
services/ | 業務邏輯 |
core/ | 共用工具（安全、例外） |
🎤 面試這樣答
Q: 你會如何組織一個 FastAPI 專案？
答案：
我會採用分層架構，將程式碼分成幾個層次：
- API 層：處理 HTTP 請求，參數驗證，呼叫服務層
- Service 層：業務邏輯，複雜的操作流程
- CRUD 層：資料庫 CRUD 操作，封裝 SQLAlchemy
- Model 層：資料庫模型（SQLAlchemy）和 Pydantic schemas
每一層只依賴下一層，不會跨層呼叫。這樣的好處是：
- 易於測試（可以 mock 下一層）
- 易於維護（修改一層不影響其他層）
- 易於擴展（新增功能只需要在對應層添加程式碼）
🤓 小測驗
Pydantic 模型應該放在哪個目錄？
資料庫 CRUD 操作應該放在哪裡？
如何避免循環依賴？
🎉 01 系列完成！
恭喜你完成 FastAPI 基礎篇！你已經學會：
- ✅ FastAPI 是什麼、為什麼選擇它
- ✅ 環境設定與第一個 API
- ✅ 路徑參數與查詢參數
- ✅ 請求體與回應模型
- ✅ 狀態碼與錯誤處理
- ✅ 自動生成 API 文件
- ✅ 專案結構最佳實踐
接下來，我們將深入學習 Pydantic 資料驗證！
上一篇： 01-6. 自動生成 API 文件 下一篇： 02-1. Pydantic 基礎模型
最後更新：2025-12-17