FastAPI 入門：打造你的第一個電子商務會員系統 API

FastAPI 入門：打造你的第一個電子商務會員系統 API

嗨，Python 新手們！準備好進入 Web 開發的世界了嗎？今天，我們要一起學習如何使用一個叫做 FastAPI 的強大框架，來搭建一個簡單而功能豐富的電子商務會員系統 API。別擔心，即使你是初次接觸 Web 框架，我也會一步一步帶你完成！

專案 GitHub 連結：https://github.com/BpsEason/ECommerce-Membership-API.git

為什麼選擇 FastAPI？

在 Python 的 Web 框架世界裡，有許多優秀的選擇，像是 Django 和 Flask。那為什麼我們選擇 FastAPI 呢？

• 速度快：FastAPI 基於 Starlette 和 Pydantic，是目前 Python 框架中最快的之一。這意味著你的 API 可以處理更多的請求，讓你的網站運行更流暢。
• 易學易用：儘管功能強大，FastAPI 的設計非常直觀，對於初學者來說非常友好。它大量利用了 Python 的型別提示（Type Hints），讓程式碼更清晰。
• 自動生成 API 文件：這是一個超級棒的功能！FastAPI 會自動為你的 API 生成互動式的 Swagger UI 和 ReDoc 文件。這表示你不需要手動編寫 API 文件，開發者和前端團隊可以直接在瀏覽器中查看和測試你的 API。
• 現代化：支援非同步程式碼 (async/await)，能更好地處理高併發請求。
• 社群活躍：有越來越多的開發者加入 FastAPI 的社群，你可以輕鬆找到資源和幫助。

專案概覽：我們要建什麼？

我們將建立一個電子商務會員系統的後端 API，它將包含以下基本功能：

• 使用者管理：
  • 註冊新會員
  • 會員登入（獲取訪問令牌）
  • 獲取個人資料
  • 更新個人資料
  • 更新密碼
  • 刪除帳戶
• 送貨地址管理：
  • 為會員添加送貨地址
  • 獲取會員的所有送貨地址
  • 獲取單一送貨地址詳情
  • 更新送貨地址
  • 刪除送貨地址

環境準備：動手前的一些設置

在開始寫程式碼之前，我們需要準備一下開發環境。

1. 安裝 Python：確保你的電腦上安裝了 Python 3.9 或更高版本。你可以從 Python 官網 下載並安裝。
2. 創建虛擬環境：為了隔離專案的依賴，我們通常會使用虛擬環境。
   Bash

   python -m venv venv
   

3. 啟動虛擬環境：
   • 在 macOS/Linux 上：
     Bash

     source venv/bin/activate
     

   • 在 Windows 上：
     Bash

     .\venv\Scripts\activate
     

   看到命令列前出現 (venv) 就表示虛擬環境已經啟動了。
4. 安裝依賴：我們需要安裝 FastAPI 和其他必要的庫。請先創建一個 requirements.txt 檔案，內容如下：

   fastapi==0.103.0
   uvicorn==0.23.2
   sqlalchemy==2.0.20
   pydantic==2.4.2
   python-jose[cryptography]==3.3.0
   passlib[bcrypt]==1.7.4
   python-dotenv==1.0.0
   # 以下為測試用，非必要但推薦
   pytest==7.4.0
   httpx==0.24.1
   

   然後安裝這些依賴：
   Bash

   pip install -r requirements.txt
   

專案結構：讓程式碼井然有序

一個好的專案結構能讓你的程式碼更容易理解和維護。我們將採用以下結構：

.
├── app/
│   ├── __init__.py
│   ├── main.py             # FastAPI 主應用程式
│   ├── database.py         # 資料庫配置
│   ├── auth.py             # 認證邏輯 (JWT, 密碼雜湊)
│   ├── models/             # SQLAlchemy 資料庫模型定義
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── address.py
│   ├── schemas/            # Pydantic 資料驗證模型 (API 輸入/輸出格式)
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── address.py
│   ├── crud/               # CRUD 操作 (與資料庫互動的邏輯)
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── address.py
│   └── routers/            # API 路由定義
│       ├── __init__.py
│       ├── users.py        # 使用者相關 API
│       └── auth.py         # 認證相關 API
├── .env                    # 環境變數文件 (本地開發用)
├── requirements.txt        # Python 依賴列表
├── README.md               # 專案說明
├── Dockerfile              # Docker 映像構建文件
└── docker-compose.yaml     # Docker Compose 配置

逐步構建你的 API

現在，讓我們一步步來編寫程式碼！

1. 配置環境變數 (.env)

在專案根目錄創建一個 .env 檔案，用於儲存敏感資訊和配置，例如資料庫連接字串和 JWT 秘密金鑰。

程式碼片段

DATABASE_URL="sqlite:///./ecommerce.db"
SECRET_KEY="your-super-secret-key" # 請將此替換為一個複雜的隨機字串！

說明：

• DATABASE_URL：這裡我們使用 SQLite，它是一個輕量級的檔案型資料庫，非常適合初學和本地開發。
• SECRET_KEY：這是用於 JWT 簽名的秘密金鑰，非常重要！ 在生產環境中，你應該使用一個真正隨機、複雜且難以猜測的字串。你可以用 python -c "import secrets; print(secrets.token_hex(32))" 生成。

2. 資料庫配置 (app/database.py)

這個檔案負責設置我們的資料庫連接。

Python

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

load_dotenv() # 載入 .env 檔案中的環境變數

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./ecommerce.db")

# 創建 SQLAlchemy 引擎
# connect_args={"check_same_thread": False} 僅用於 SQLite
engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {}
)

# 創建會話工廠，每次請求時將從這裡獲取資料庫會話
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 基礎模型類，我們的所有資料庫模型都將繼承它
Base = declarative_base()

def get_db():
    """獲取資料庫會話，確保使用後關閉"""
    db = SessionLocal()
    try:
        yield db # 使用 yield 確保會話在請求完成後自動關閉
    finally:
        db.close()

說明：

• create_engine：建立資料庫連接的引擎。
• SessionLocal：用於創建資料庫會話的類別。每個 API 請求都將有一個獨立的資料庫會話。
• Base = declarative_base()：所有 SQLAlchemy 模型都會繼承這個 Base。
• get_db()：這是一個 FastAPI 的依賴注入（Dependency Injection）函數。FastAPI 會自動調用它來為每個請求提供一個資料庫會話，並在請求結束後負責關閉它。

3. 資料庫模型 (app/models/user.py & app/models/address.py)

這些檔案定義了資料庫中的表結構。

app/models/user.py

Python

from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from ..database import Base # 注意路徑，表示上一級目錄的 database.py

class User(Base):
    __tablename__ = "users" # 資料庫中的表名

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String, nullable=True) # 新增：真實姓名
    phone_number = Column(String, unique=True, index=True, nullable=True) # 新增：聯絡電話
    is_active = Column(Boolean, default=True) # 新增：帳戶是否啟用
    is_verified = Column(Boolean, default=False) # 新增：電子郵件是否驗證
    created_at = Column(DateTime, server_default=func.now()) # 自動設定創建時間
    updated_at = Column(DateTime, onupdate=func.now(), server_default=func.now()) # 自動更新修改時間
    last_login_at = Column(DateTime, nullable=True) # 新增：上次登入時間

    # 定義與 Address 的一對多關係，一個使用者可以有多個地址
    # cascade="all, delete-orphan" 表示刪除使用者時，其所有地址也會被刪除
    addresses = relationship("Address", back_populates="user", cascade="all, delete-orphan")

    def __repr__(self): # 方便在調試時查看對象資訊
        return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"

app/models/address.py

Python

from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from ..database import Base # 注意路徑

class Address(Base):
    __tablename__ = "addresses" # 資料庫中的表名

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False) # 外鍵，關聯到 users 表的 id
    address_line1 = Column(String, nullable=False)
    address_line2 = Column(String, nullable=True)
    city = Column(String, nullable=False)
    state_province = Column(String, nullable=False)
    zip_code = Column(String, nullable=False)
    country = Column(String, nullable=False)
    is_default = Column(Boolean, default=False) # 是否為預設地址
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, onupdate=func.now(), server_default=func.now())

    # 定義與 User 的多對一關係，一個地址只屬於一個使用者
    user = relationship("User", back_populates="addresses")

    def __repr__(self):
        return f"<Address(id={self.id}, user_id={self.user_id}, city='{self.city}')>"

說明：

• 我們使用 SQLAlchemy 來定義資料庫模型。每個類別（User, Address）都對應資料庫中的一張表。
• Column 定義了表的欄位，指定了資料型別（Integer, String, Boolean 等）和屬性（primary_key, unique, nullable）。
• ForeignKey 用於建立表之間的關係（例如 Address 表的 user_id 欄位指向 User 表的 id 欄位）。
• relationship 允許你方便地在 Python 物件之間導航，例如通過 user.addresses 獲取一個使用者的所有地址。

4. 資料驗證模型 (app/schemas/user.py & app/schemas/address.py)

這些檔案定義了 API 請求和回應的資料格式，使用 Pydantic 進行資料驗證。

app/schemas/user.py

Python

from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional, List
from .address import Address # 導入同級目錄下的 Address Schema

# 用於創建或更新使用者的基礎資訊
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, description="使用者名稱")
    email: EmailStr = Field(..., description="電子郵件")
    full_name: Optional[str] = Field(None, max_length=100, description="真實姓名")
    phone_number: Optional[str] = Field(None, max_length=20, description="聯絡電話")

# 用於使用者註冊 (包含密碼)
class UserCreate(UserBase):
    password: str = Field(..., min_length=6, description="密碼")

# 用於更新使用者資訊 (所有欄位可選)
class UserUpdate(UserBase):
    username: Optional[str] = None
    email: Optional[EmailStr] = None
    full_name: Optional[str] = None
    phone_number: Optional[str] = None
    is_active: Optional[bool] = None # 帳戶狀態，通常由管理員修改
    is_verified: Optional[bool] = None # 電子郵件驗證狀態

# 用於更新密碼
class UserPasswordUpdate(BaseModel):
    old_password: str = Field(..., description="舊密碼")
    new_password: str = Field(..., min_length=6, description="新密碼")

# 用於 API 回應的使用者模型 (包含資料庫生成的資訊和關聯地址)
class User(UserBase):
    id: int
    is_active: bool
    is_verified: bool
    created_at: datetime
    updated_at: datetime
    last_login_at: Optional[datetime]
    addresses: List[Address] = [] # 包含地址列表

    class Config: # 允許從 SQLAlchemy ORM 物件創建 Pydantic 模型
        from_attributes = True

# 用於使用者登入
class UserLogin(BaseModel):
    username: str # 可以是 username 或 email，根據實際登入邏輯決定
    password: str

# JWT Token 回應模型
class Token(BaseModel):
    access_token: str
    token_type: str

# JWT Token 內部的資料模型 (用於解碼 token)
class TokenData(BaseModel):
    username: Optional[str] = None

# 密碼重設請求 (忘記密碼)
class PasswordResetRequest(BaseModel):
    email: EmailStr

# 密碼重設確認
class PasswordResetConfirm(BaseModel):
    token: str # 從郵件連結中獲取的 token
    new_password: str

app/schemas/address.py

Python

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

# 用於創建或更新地址的基礎資訊
class AddressBase(BaseModel):
    address_line1: str = Field(..., description="地址第一行")
    address_line2: Optional[str] = Field(None, description="地址第二行")
    city: str = Field(..., description="城市")
    state_province: str = Field(..., description="省/州")
    zip_code: str = Field(..., description="郵遞區號")
    country: str = Field(..., description="國家")
    is_default: bool = Field(False, description="是否為預設地址")

# 用於創建地址
class AddressCreate(AddressBase):
    pass

# 用於更新地址 (所有欄位可選)
class AddressUpdate(AddressBase):
    address_line1: Optional[str] = None
    address_line2: Optional[str] = None
    city: Optional[str] = None
    state_province: Optional[str] = None
    zip_code: Optional[str] = None
    country: Optional[str] = None
    is_default: Optional[bool] = None

# 用於 API 回應的地址模型 (包含資料庫生成的資訊)
class Address(AddressBase):
    id: int
    user_id: int
    created_at: datetime
    updated_at: datetime

    class Config: # 允許從 SQLAlchemy ORM 物件創建 Pydantic 模型
        from_attributes = True

說明：

• Pydantic 負責 API 請求和回應的資料驗證和序列化。
• BaseModel 是所有 Pydantic 模型的基類。
• Field 用於添加額外的驗證規則（如 min_length, max_length）和描述。
• Optional 說明欄位是可選的。
• Config.from_attributes = True 是 FastAPI 與 SQLAlchemy 協同工作的關鍵，它允許 Pydantic 模型直接從 ORM 物件讀取屬性。

5. CRUD 操作 (app/crud/user.py & app/crud/address.py)

這些檔案包含與資料庫互動的實際邏輯（創建 Create, 讀取 Read, 更新 Update, 刪除 Delete）。

app/crud/user.py

Python

from sqlalchemy.orm import Session, joinedload # joinedload 用於優化關聯查詢
from ..models.user import User # 導入 SQLAlchemy User 模型
from ..schemas.user import UserCreate, UserUpdate # 導入 Pydantic Schema
from ..auth import get_password_hash # 導入密碼雜湊函數
from datetime import datetime

def get_user(db: Session, user_id: int):
    """根據使用者 ID 獲取使用者，包含地址 (預先載入)"""
    # 使用 joinedload 預先載入 addresses，避免 N+1 問題 (多次查詢資料庫)
    return db.query(User).options(joinedload(User.addresses)).filter(User.id == user_id).first()

def get_user_by_username(db: Session, username: str):
    """根據使用者名稱獲取使用者"""
    return db.query(User).filter(User.username == username).first()

def get_user_by_email(db: Session, email: str):
    """根據電子郵件獲取使用者"""
    return db.query(User).filter(User.email == email).first()

def create_user(db: Session, user: UserCreate):
    """創建新使用者"""
    hashed_password = get_password_hash(user.password) # 雜湊密碼
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password,
        full_name=user.full_name,
        phone_number=user.phone_number
    )
    db.add(db_user) # 添加到會話
    db.commit() # 提交到資料庫
    db.refresh(db_user) # 刷新物件以獲取資料庫生成的 ID 等資訊
    return db_user

def update_user(db: Session, user_id: int, user_update: UserUpdate):
    """更新使用者資訊"""
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        # dict(exclude_unset=True) 只包含 Pydantic 模型中已設置的欄位
        update_data = user_update.dict(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_user, key, value) # 更新模型屬性
        db.commit()
        db.refresh(db_user)
    return db_user

def update_user_password(db: Session, user_id: int, hashed_password: str):
    """更新使用者密碼"""
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        db_user.hashed_password = hashed_password
        db.commit()
        db.refresh(db_user)
    return db_user

def update_user_last_login(db: Session, user_id: int):
    """更新使用者上次登入時間"""
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        db_user.last_login_at = datetime.utcnow() # 使用 UTC 時間
        db.commit()
        db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    """刪除使用者"""
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        db.delete(db_user) # 從會話中刪除
        db.commit() # 提交到資料庫
    return db_user

app/crud/address.py

Python

from sqlalchemy.orm import Session
from ..models.address import Address # 導入 SQLAlchemy Address 模型
from ..schemas.address import AddressCreate, AddressUpdate # 導入 Pydantic Schema

def get_user_addresses(db: Session, user_id: int):
    """獲取指定使用者的所有送貨地址"""
    return db.query(Address).filter(Address.user_id == user_id).all()

def get_address(db: Session, address_id: int):
    """根據地址 ID 獲取地址"""
    return db.query(Address).filter(Address.id == address_id).first()

def create_address(db: Session, address: AddressCreate, user_id: int):
    """創建新送貨地址"""
    # 如果要創建的地址是預設地址，則將該使用者原有的預設地址設為非預設
    if address.is_default:
        current_default = db.query(Address).filter(Address.user_id == user_id, Address.is_default == True).first()
        if current_default:
            current_default.is_default = False
            db.add(current_default) # 標記為需要更新

    db_address = Address(**address.dict(), user_id=user_id) # 將 Pydantic 模型轉換為 SQLAlchemy 模型
    db.add(db_address)
    db.commit()
    db.refresh(db_address)
    return db_address

def update_address(db: Session, address_id: int, address_update: AddressUpdate, user_id: int):
    """更新送貨地址"""
    db_address = db.query(Address).filter(Address.id == address_id, Address.user_id == user_id).first()
    if db_address:
        update_data = address_update.dict(exclude_unset=True)
        # 如果嘗試將此地址設為預設，則取消其他預設地址
        if 'is_default' in update_data and update_data['is_default'] == True:
            current_default = db.query(Address).filter(Address.user_id == user_id, Address.is_default == True).first()
            if current_default and current_default.id != address_id: # 確保不是更新自己為預設
                current_default.is_default = False
                db.add(current_default)

        for key, value in update_data.items():
            setattr(db_address, key, value)
        db.commit()
        db.refresh(db_address)
    return db_address

def delete_address(db: Session, address_id: int, user_id: int):
    """刪除送貨地址"""
    db_address = db.query(Address).filter(Address.id == address_id, Address.user_id == user_id).first()
    if db_address:
        db.delete(db_address)
        db.commit()
    return db_address

說明：

• CRUD 函數封裝了所有與資料庫的交互，保持了 API 路由的簡潔。
• 它們接收 db: Session 參數，這是通過 FastAPI 依賴注入提供的。
• joinedload 是一個優化技巧，用於在查詢 User 時一併載入相關的 Address 資料，避免了「N+1 查詢問題」（每次查詢一個使用者都發送額外的查詢去獲取其地址）。
• create_address 和 update_address 中包含了處理預設地址的邏輯，確保一個使用者只有一個預設地址。

6. 認證邏輯 (app/auth.py)

這個檔案處理密碼雜湊、JWT 生成和驗證，是整個系統的安全核心。

Python

from fastapi import HTTPException, status, Depends
from jose import JWTError, jwt # 用於 JWT 的生成和驗證
from passlib.context import CryptContext # 用於密碼雜湊
from datetime import datetime, timedelta
import os
from dotenv import load_dotenv
from sqlalchemy.orm import Session
from ..database import get_db # 導入資料庫會話
from ..crud.user import get_user_by_username # 導入獲取使用者函數
from ..models.user import User as DBUser # 導入 SQLAlchemy User 模型

load_dotenv() # 載入 .env 檔案

# JWT 配置
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key") # 從環境變數獲取秘密金鑰
ALGORITHM = "HS256" # JWT 簽名演算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 訪問令牌的過期時間 (分鐘)

# 密碼雜湊上下文，使用 bcrypt 演算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """驗證明文密碼是否與雜湊密碼匹配"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """生成密碼的雜湊值"""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    """生成 JWT 訪問令牌"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire}) # 將過期時間添加到 JWT payload
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) # 簽名生成 JWT
    return encoded_jwt

# 這是一個依賴注入函數，用於獲取當前使用者
# 它會從請求頭部的 Authorization: Bearer <token> 中解析 JWT
async def get_current_user(token: str, db: Session = Depends(get_db)) -> DBUser:
    """從 JWT 獲取當前使用者對象"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="無法驗證憑證",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) # 解碼 JWT
        username: str = payload.get("sub") # 獲取使用者名稱
        if username is None:
            raise credentials_exception
    except JWTError: # 如果 JWT 無效或過期
        raise credentials_exception
    
    # 根據使用者名稱從資料庫中獲取完整的 User 對象
    user = get_user_by_username(db, username=username)
    if user is None:
        raise credentials_exception
    return user # 返回使用者對象

說明：

• passlib 用於安全地雜湊和驗證密碼。
• python-jose 用於處理 JWT（JSON Web Token），這是一種安全的身份驗證方式。
• get_current_user 是一個非常重要的依賴函數，它會驗證每個受保護的 API 請求中的 JWT，並返回當前登入的使用者對象。如果驗證失敗，它會拋出 401 未授權錯誤。

7. API 路由 (app/routers/auth.py & app/routers/users.py)

這些檔案定義了你的 API 端點（URL）。

app/routers/auth.py

Python

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm # 用於處理 OAuth2 密碼流程的表單
from sqlalchemy.orm import Session
from datetime import timedelta
from ..database import get_db # 導入資料庫會話
from ..schemas.user import UserCreate, User, Token, PasswordResetRequest, PasswordResetConfirm # 導入 Pydantic Schema
from ..crud.user import get_user_by_username, get_user_by_email, create_user, update_user_last_login, update_user_password, get_user
from ..auth import verify_password, create_access_token, ACCESS_TOKEN_EXPIRE_MINUTES, get_password_hash
from ..models.user import User as DBUser # 導入 SQLAlchemy User 模型

router = APIRouter()

@router.post("/register", response_model=User, status_code=status.HTTP_201_CREATED, summary="註冊新會員")
async def register_user(user: UserCreate, db: Session = Depends(get_db)):
    """
    註冊新會員帳戶。
    - 檢查使用者名稱和電子郵件是否已被註冊。
    - 創建新使用者並返回其資訊。
    """
    db_user_by_username = get_user_by_username(db, username=user.username)
    if db_user_by_username:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="使用者名稱已被註冊")
    db_user_by_email = get_user_by_email(db, email=user.email)
    if db_user_by_email:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="電子郵件已被註冊")
    return create_user(db=db, user=user)

@router.post("/token", response_model=Token, summary="會員登入並獲取訪問令牌")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    """
    使用者登入，成功後返回 JWT 訪問令牌。
    - 支援使用使用者名稱或電子郵件登入 (OAuth2PasswordRequestForm 預設是 username)。
    - 驗證密碼。
    - 更新上次登入時間。
    """
    # 這裡 form_data.username 可能是 email 或 username，視實際登入策略而定
    # 這裡我們簡化，假設它直接是 username
    user = get_user_by_username(db, username=form_data.username) 
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="不正確的使用者名稱或密碼",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 登入成功後更新上次登入時間
    update_user_last_login(db, user.id)

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# --- 忘記密碼/密碼重設 (僅為示意，需要額外 Token 管理和郵件發送服務) ---
# 實際應用中需要額外的 Token 管理（例如獨立的重設密碼Token，而非JWT）
# 並且需要結合郵件發送服務
import jwt # 假設你需要手動導入 jwt，如果已在 auth.py 導入則不需要

@router.post("/forgot-password", summary="發送密碼重設郵件")
async def forgot_password(request: PasswordResetRequest, db: Session = Depends(get_db)):
    """
    使用者忘記密碼，發送重設密碼郵件。
    （這裡僅為示意，不包含實際郵件發送邏輯）
    """
    user = get_user_by_email(db, email=request.email)
    if not user:
        # 為了安全，即使電子郵件不存在，也返回成功訊息，避免洩漏使用者資訊
        return {"message": "如果此電子郵件已註冊，重設密碼連結將發送到您的信箱。"}

    # 實際應用中：
    # 1. 生成一個獨立的、帶有有效期限的重設密碼Token (與 JWT 不同)
    # 2. 將此 Token 儲存到資料庫中，與使用者關聯
    # 3. 發送包含此 Token 的重設連結到使用者郵箱
    # 例如：reset_token = create_reset_password_token(user.id)
    # send_email_with_reset_link(user.email, reset_token)

    return {"message": "如果此電子郵件已註冊，重設密碼連結將發送到您的信箱。"}

@router.post("/reset-password", summary="重設密碼")
async def reset_password(request: PasswordResetConfirm, db: Session = Depends(get_db)):
    """
    使用者提交重設密碼表單，包含從郵件中獲取的 Token 和新密碼。
    （這裡僅為示意，不包含實際 Token 驗證邏輯）
    """
    # 實際應用中：
    # 1. 驗證 request.token 的有效性，包括是否過期，以及是否與使用者匹配
    # 2. 獲取 Token 關聯的使用者 ID

    # 為了範例，我們簡化並假設 token 包含 user_id
    try:
        # 注意：這裡的 SECRET_KEY 應該是專用於重設密碼 Token 的，與 JWT 不同
        # 且實際 payload 應該包含 user_id，而不是直接讓 user_id = payload.get("sub")
        # 這裡僅為範例目的，直接假設 token 解碼後能得到 user_id
        payload = jwt.decode(request.token, SECRET_KEY, algorithms=["HS256"]) 
        user_id = payload.get("sub") # 假設 'sub' 包含 user_id
        if user_id is None:
            raise ValueError
    except (JWTError, ValueError):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="無效或過期的重設密碼連結",
        )

    user = get_user(db, user_id=int(user_id))
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="使用者未找到")

    new_hashed_password = get_password_hash(request.new_password)
    update_user_password(db, user.id, new_hashed_password)

    return {"message": "密碼已成功重設。"}

app/routers/users.py

Python

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from ..database import get_db # 導入資料庫會話
from ..schemas.user import User, UserUpdate, UserPasswordUpdate # 導入 Pydantic User Schema
from ..schemas.address import Address, AddressCreate, AddressUpdate # 導入 Pydantic Address Schema
from ..crud.user import update_user, update_user_password, delete_user # 導入使用者 CRUD 函數
from ..crud.address import get_user_addresses, get_address, create_address, update_address, delete_address # 導入地址 CRUD 函數
from ..auth import get_current_user, verify_password, get_password_hash # 導入認證相關函數
from ..models.user import User as DBUser # 導入 SQLAlchemy User 模型，為了型別提示

router = APIRouter()

# 依賴注入函數：獲取當前活躍使用者
# 注意：這裡的 current_user 直接就是 DBUser 物件了，不再是字串
async def get_current_active_user(current_user: DBUser = Depends(get_current_user)):
    """
    獲取當前登入且活躍的使用者。
    如果使用者帳戶未啟用，則拋出未授權錯誤。
    """
    if not current_user.is_active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="使用者帳戶未啟用")
    return current_user

# --- 使用者個人資訊管理 ---

@router.get("/me", response_model=User, summary="獲取當前使用者資訊和地址")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    """
    獲取當前登入使用者的詳細資訊，包含其所有送貨地址。
    此端點需要認證。
    """
    # current_user 已經是完整的 DBUser 對象，FastAPI 會自動將其轉換為 Pydantic User 模型
    return current_user

@router.put("/me", response_model=User, summary="更新當前使用者個人資訊")
async def update_users_me(user_update: UserUpdate,
                          current_user: DBUser = Depends(get_current_active_user),
                          db: Session = Depends(get_db)):
    """
    更新當前登入使用者的個人資訊（例如：真實姓名、電話號碼）。
    - 密碼更新請使用 `/me/password` 端點。
    - 此端點需要認證。
    """
    updated_user = update_user(db, current_user.id, user_update)
    if not updated_user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="使用者未找到")
    return updated_user

@router.put("/me/password", response_model=User, summary="更新當前使用者密碼")
async def update_my_password(password_update: UserPasswordUpdate,
                            current_user: DBUser = Depends(get_current_active_user),
                            db: Session = Depends(get_db)):
    """
    更新當前登入使用者的密碼。
    - 需要提供舊密碼進行驗證。
    - 此端點需要認證。
    """
    if not verify_password(password_update.old_password, current_user.hashed_password):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="舊密碼不正確")

    new_hashed_password = get_password_hash(password_update.new_password)
    updated_user = update_user_password(db, current_user.id, new_hashed_password)
    if not updated_user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="使用者未找到")
    return updated_user

@router.delete("/me", status_code=status.HTTP_204_NO_CONTENT, summary="刪除當前使用者帳戶")
async def delete_users_me(current_user: DBUser = Depends(get_current_active_user),
                          db: Session = Depends(get_db)):
    """
    刪除當前登入使用者的帳戶。
    - 此操作不可逆。
    - 此端點需要認證。
    """
    deleted = delete_user(db, current_user.id)
    if not deleted:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="使用者未找到")
    return {"message": "帳戶已成功刪除"}

# --- 地址管理 ---

@router.get("/me/addresses", response_model=List[Address], summary="獲取當前使用者所有送貨地址")
async def read_my_addresses(current_user: DBUser = Depends(get_current_active_user),
                            db: Session = Depends(get_db)):
    """
    獲取當前登入使用者的所有送貨地址列表。
    此端點需要認證。
    """
    addresses = get_user_addresses(db, current_user.id)
    return addresses

@router.post("/me/addresses", response_model=Address, status_code=status.HTTP_201_CREATED, summary="為當前使用者新增送貨地址")
async def create_my_address(address: AddressCreate,
                            current_user: DBUser = Depends(get_current_active_user),
                            db: Session = Depends(get_db)):
    """
    為當前登入使用者新增一個送貨地址。
    - 可設定為預設地址。
    - 此端點需要認證。
    """
    return create_address(db, address, current_user.id)

@router.get("/me/addresses/{address_id}", response_model=Address, summary="獲取當前使用者指定送貨地址")
async def read_my_address(address_id: int,
                          current_user: DBUser = Depends(get_current_active_user),
                          db: Session = Depends(get_db)):
    """
    獲取當前登入使用者指定 ID 的送貨地址詳情。
    - 確保地址屬於當前使用者。
    - 此端點需要認證。
    """
    address = get_address(db, address_id)
    if not address or address.user_id != current_user.id:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="地址未找到或不屬於此使用者")
    return address

@router.put("/me/addresses/{address_id}", response_model=Address, summary="更新當前使用者指定送貨地址")
async def update_my_address(address_id: int,
                            address_update: AddressUpdate,
                            current_user: DBUser = Depends(get_current_active_user),
                            db: Session = Depends(get_db)):
    """
    更新當前登入使用者指定 ID 的送貨地址。
    - 確保地址屬於當前使用者。
    - 可更新為預設地址，將自動取消其他預設地址。
    - 此端點需要認證。
    """
    updated_address = update_address(db, address_id, address_update, current_user.id)
    if not updated_address:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="地址未找到或不屬於此使用者")
    return updated_address

@router.delete("/me/addresses/{address_id}", status_code=status.HTTP_204_NO_CONTENT, summary="刪除當前使用者指定送貨地址")
async def delete_my_address(address_id: int,
                            current_user: DBUser = Depends(get_current_active_user),
                            db: Session = Depends(get_db)):
    """
    刪除當前登入使用者指定 ID 的送貨地址。
    - 確保地址屬於當前使用者。
    - 此操作不可逆。
    - 此端點需要認證。
    """
    deleted = delete_address(db, address_id, current_user.id)
    if not deleted:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="地址未找到或不屬於此使用者")
    return {"message": "地址已成功刪除"}

# --- 管理員專用端點 (需要額外權限檢查) ---
# 假設我們有一個簡單的isAdmin的檢查函數
# 實際應用中需要根據你的實際權限系統來判斷
# 例如：可以增加一個 is_admin 欄位到 User 模型，或使用角色權限系統
def is_admin(current_user: DBUser = Depends(get_current_active_user)):
    # 這裡你需要替換為你實際的管理員判斷邏輯
    # 例如：if not current_user.is_admin:
    #     raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="無權訪問")
    pass # 目前為空，實際使用需要添加邏輯

@router.get("/", response_model=List[User], summary="獲取所有使用者 (僅限管理員)")
async def read_all_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db),
                         admin_check: None = Depends(is_admin)): # 添加管理員檢查
    """
    獲取所有使用者資訊列表。
    此端點僅限管理員訪問。
    """
    users = db.query(DBUser).offset(skip).limit(limit).all()
    return users

說明：

• APIRouter() 用於將相關的 API 端點分組，並在 main.py 中進行註冊。
• 每個 @router.xxx 裝飾器定義了一個 HTTP 方法（GET, POST, PUT, DELETE）和一個路徑。
• response_model 告訴 FastAPI API 回應的 Pydantic 模型，FastAPI 會自動進行序列化。
• status_code 定義了回應的 HTTP 狀態碼。
• summary 和 docstring 提供了 API 的簡要說明和詳細描述，這些會自動呈現在 Swagger UI 中。
• Depends(get_db) 注入了資料庫會話。
• Depends(get_current_active_user) 用於保護端點，確保只有登入且活躍的使用者才能訪問。

8. 主應用程式 (app/main.py)

這是 FastAPI 應用程式的入口點。

Python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware # 用於處理跨域請求
from .routers import users, auth # 導入路由
from .database import engine # 導入資料庫引擎
from .models.user import User # 導入 SQLAlchemy User 模型
from .models.address import Address # 導入 SQLAlchemy Address 模型

app = FastAPI(
    title="電子商務會員系統",
    description="基於 FastAPI 的可擴展電子商務會員系統 API，具備使用者和地址管理功能。",
    version="1.0.0"
)

# 創建資料庫表 (如果不存在)。這會在應用程式啟動時執行。
User.metadata.create_all(bind=engine)
Address.metadata.create_all(bind=engine)

# CORS 中介軟體配置
# 允許來自所有來源的跨域請求 (生產環境應限制特定域名以增強安全性)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 實際部署時請替換為前端的域名，例如 ["http://localhost:3000", "https://yourfrontend.com"]
    allow_credentials=True, # 允許傳送 cookie
    allow_methods=["*"], # 允許所有 HTTP 方法
    allow_headers=["*"], # 允許所有 HTTP 頭部
)

# 包含 API 路由
app.include_router(auth.router, prefix="/auth", tags=["認證"]) # 認證相關路由，前綴 /auth
app.include_router(users.router, prefix="/users", tags=["使用者"]) # 使用者相關路由，前綴 /users

@app.get("/", summary="根端點 (健康檢查)")
async def root():
    """根端點，用於健康檢查，返回應用程式基本資訊。"""
    return {"message": "電子商務會員系統 API"}

說明：

• FastAPI() 初始化了應用程式。
• User.metadata.create_all(bind=engine) 和 Address.metadata.create_all(bind=engine) 會根據你定義的 SQLAlchemy 模型自動創建資料庫表。
• CORSMiddleware 處理跨域請求，這在前後端分離的專案中非常常見。
• app.include_router 將之前定義的路由模組註冊到主應用程式中。
• 根路徑 / 是一個簡單的健康檢查點。

9. 啟動你的 API

在專案根目錄下（即 main.py 所在的 app 資料夾的上一級），確保你已啟動虛擬環境，然後運行：

Bash

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

• uvicorn: FastAPI 的推薦 ASGI 伺服器。
• app.main:app: 指定你的應用程式實例在哪裡 (app 資料夾裡的 main.py 檔案裡的 app 實例)。
• --host 0.0.0.0: 讓伺服器監聽所有可用的網路接口。
• --port 8000: 讓伺服器監聽 8000 埠。
• --reload: 在程式碼變更時自動重載伺服器 (開發時非常有用)。

打開你的瀏覽器，訪問 http://localhost:8000/docs，你將看到自動生成的 Swagger UI 互動式 API 文件！你可以在這裡測試你的每一個 API 端點。

Docker 部署 (可選，但推薦學習)

如果你想將你的應用程式打包成 Docker 容器，你已經有了 Dockerfile 和 docker-compose.yaml。

Dockerfile

Dockerfile

# 使用輕量級的 Python 3.9 基礎映像
FROM python:3.9-slim

# 設定容器內的工作目錄
WORKDIR /app

# 將 requirements.txt 複製到工作目錄，並安裝依賴
# 這樣做可以利用 Docker 層的快取，如果 requirements.txt 沒有變化，則無需重新安裝依賴
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 將所有專案檔案複製到容器內的工作目錄
COPY . .

# 定義容器啟動時執行的命令，使用 uvicorn 啟動 FastAPI 應用
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yaml

YAML

version: '3.8' # Docker Compose 文件版本

services:
  app: # 定義一個名為 'app' 的服務
    build: . # 從當前目錄的 Dockerfile 構建映像
    ports:
      - "8000:8000" # 將主機的 8000 埠映射到容器的 8000 埠
    environment: # 設定容器內部的環境變數
      - DATABASE_URL=sqlite:///./ecommerce.db # 使用 SQLite 資料庫
      - SECRET_KEY=your-secret-key # 請替換為安全的秘密金鑰！
    volumes: # 掛載卷，方便開發時同步程式碼變更
      - .:/app # 將主機的當前目錄掛載到容器的 /app

部署步驟：

1. 確保你已經安裝了 Docker Desktop。
2. 在專案根目錄（包含 Dockerfile 和 docker-compose.yaml 的目錄）打開終端機。
3. 運行命令構建並啟動容器：
   Bash

   docker-compose up --build -d
   

   • --build: 如果映像不存在或 Dockerfile 有修改，則重新構建。
   • -d: 在後台運行容器。
4. 訪問 http://localhost:8000/docs 即可。
5. 停止容器：
   Bash

   docker-compose down
   

未來可以擴展的功能：

• 電子郵件驗證：在使用者註冊後發送確認郵件，確保電子郵件的真實性，並將 is_verified 設置為 True。
• 密碼重設：實現一個完整的忘記密碼流程，包括生成一次性重設連結、發送郵件和驗證 Token。
• 會員等級與積分：設計會員等級制度，根據用戶的消費或其他行為給予積分，並自動升級。
• 購物車與訂單管理：這是電子商務的核心，可以擴展資料庫模型和 API 來處理購物車商品、創建訂單、管理訂單狀態等。
• 商品資訊整合：如果有的話，與商品資料庫整合，讓會員系統能查詢商品資訊。
• 支付整合：與第三方支付閘道器（如 Stripe, PayPal）整合。
• 後台管理介面：為管理員提供一個介面來管理使用者、訂單等。

結語

恭喜你！你已經完成了使用 FastAPI 搭建第一個電子商務會員系統 API 的基本框架。這是一個非常好的開始，你可以根據這個基礎，不斷學習和擴展，打造更複雜、更實用的應用程式。Web 開發充滿樂趣，希望你享受這個過程！