# 

# 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 - 應用程式入口

```python
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 - 設定管理

```python
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 - 共用依賴

```python
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 - 路由聚合

```python
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 - 使用者路由

```python
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 user
```

### app/schemas/user.py - Pydantic 模型

```python
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

```python
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. 命名規範

```python
# 檔案命名：小寫 + 底線
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 |
| 一般應用 | 基本分層 |
| 生產環境 | 完整分層 + 測試 |

### 核心原則

1. **關注點分離** - 每個模組做一件事
2. **依賴注入** - 使用 FastAPI 的 Depends
3. **統一命名** - 遵循 Python 命名規範
4. **易於測試** - 業務邏輯與框架分離

### 檔案職責

| 檔案/目錄 | 職責 |
|----------|------|
| `main.py` | 應用程式建立和設定 |
| `config.py` | 環境變數和設定 |
| `api/` | HTTP 路由處理 |
| `schemas/` | 請求/回應模型 |
| `models/` | 資料庫模型 |
| `crud/` | 資料存取層 |
| `services/` | 業務邏輯 |
| `core/` | 共用工具（安全、例外） |

---

## 🎤 面試這樣答

### Q: 你會如何組織一個 FastAPI 專案？

**答案：**

> 我會採用分層架構，將程式碼分成幾個層次：
>
> 1. **API 層**：處理 HTTP 請求，參數驗證，呼叫服務層
> 2. **Service 層**：業務邏輯，複雜的操作流程
> 3. **CRUD 層**：資料庫 CRUD 操作，封裝 SQLAlchemy
> 4. **Model 層**：資料庫模型（SQLAlchemy）和 Pydantic schemas
>
> 每一層只依賴下一層，不會跨層呼叫。這樣的好處是：
> - 易於測試（可以 mock 下一層）
> - 易於維護（修改一層不影響其他層）
> - 易於擴展（新增功能只需要在對應層添加程式碼）

---

## 🤓 小測驗

1. Pydantic 模型應該放在哪個目錄？
   <details>
   <summary>點擊看答案</summary>
   schemas/ 目錄
   </details>

2. 資料庫 CRUD 操作應該放在哪裡？
   <details>
   <summary>點擊看答案</summary>
   crud/ 目錄
   </details>

3. 如何避免循環依賴？
   <details>
   <summary>點擊看答案</summary>
   保持依賴方向單向：api → services → crud → models
   </details>

---

## 🎉 01 系列完成！

恭喜你完成 FastAPI 基礎篇！你已經學會：

- ✅ FastAPI 是什麼、為什麼選擇它
- ✅ 環境設定與第一個 API
- ✅ 路徑參數與查詢參數
- ✅ 請求體與回應模型
- ✅ 狀態碼與錯誤處理
- ✅ 自動生成 API 文件
- ✅ 專案結構最佳實踐

接下來，我們將深入學習 Pydantic 資料驗證！

---

**上一篇：** [01-6. 自動生成 API 文件](./01-6)
**下一篇：** [02-1. Pydantic 基礎模型](./02-1)

---

最後更新：2025-12-17


---

> 作者: luk  
> URL: https://yoru-karu-blog-lalaluk-52581ac5e0cef170a3c8922c19182ecb6f7bd604.gitlab.io/posts/tutorial/fastapi/01-7/