余小章 @ 大內殿堂

祕訣無它，唯勤而已;唯有不斷學習，才能成長 by 余小章

2025-05-04

使用 pytest 測試 API (FastAPI) - 續

上篇  使用 Memory 作為 Web API 的 Repository，這次改用 Postgresql 並搭配 DI Container，也包含了資料庫的操作驗證

開發環境設置

  • Windows 11 Home
  • Python 3.13
  • PyCharm
  • uv

先建立虛擬環境

uv venv --seed --link-mode=copy

安裝必要的套件

通過 uv 安裝

uv add "dependency-injector>=4.46.0" "dotenv>=0.9.9" "fastapi>=0.115.12" "httpx>=0.28.1" "psycopg2>=2.9.10" "pydantic>=2.11.2" "pytest>=8.3.5" "python-dateutil>=2.9.0.post0" "sqlalchemy>=2.0.40" "testcontainers-postgres>=0.0.1rc1" "testcontainers-redis>=0.0.1rc1" "uvicorn>=0.34.0"

 

或者，使用 uv pip install -e . 還原套件，範例專案已存在 pyproject.toml，內容如下

[project]
name = "lab-py-apitest"
version = "0.1.0"
description = "Add your description here"
requires-python = ">=3.13"
dependencies = [
    "dependency-injector>=4.46.0",
    "dotenv>=0.9.9",
    "fastapi>=0.115.12",
    "httpx>=0.28.1",
    "psycopg2>=2.9.10",
    "pydantic>=2.11.2",
    "pytest>=8.3.5",
    "python-dateutil>=2.9.0.post0",
    "sqlalchemy>=2.0.40",
    "testcontainers-postgres>=0.0.1rc1",
    "testcontainers-redis>=0.0.1rc1",
    "uvicorn>=0.34.0",
]

[tool.setuptools]
packages = ["app", "tests"]

 

還原依賴套件

uv pip install -e . --link-mode=copy

 

實作儲存庫模式 (Repository Pattern) 

儲存庫模式是一種設計模式，它將資料存取邏輯與業務邏輯分離，使我們可以輕鬆地切換不同的資料存儲實現，而不影響業務邏輯。目前這個範例並沒有業務邏輯，只有單純的資料存取，原則上，存取邏輯是設計給業務邏輯使用。

定義儲存庫介面

首先，我們定義一個抽象基類作為儲存庫介面

# app/db/member_repository.py

from abc import ABC, abstractmethod
from typing import List, Optional
from app.models.member import Member, MemberCreate, UpdateMemberRequest
class MemberRepositoryInterface(ABC):
   @abstractmethod
   def get_all(self) -> List[Member]:
       pass
   @abstractmethod
   def get_by_id(self, member_id: str) -> Optional[Member]:
       pass
   @abstractmethod
   def create(self, member_create: MemberCreate) -> Member:
       pass
   @abstractmethod
   def update(self, member_id: str, update_member_request: UpdateMemberRequest) -> Optional[Member]:
       pass
   @abstractmethod
   def delete(self, member_id: str) -> bool:
       pass

 

實現記憶體的儲存庫

# app/db/member_memory_repository.py

from typing import Dict, List, Optional

from app.db.member_repository import MemberRepositoryInterface
from app.models.member import Member, MemberCreate, UpdateMemberRequest

class MemoryMemberRepository(MemberRepositoryInterface):
   def __init__(self):
       self.members: Dict[str, Member] = {}

   def get_all(self) -> List[Member]:
       return list(self.members.values())

   def get_by_id(self, member_id: str) -> Optional[Member]:
       return self.members.get(member_id)

   def create(self, member_create: MemberCreate) -> Member:
       dump = member_create.model_dump()
       member = Member(**dump)
       self.members[member.id] = member
       return member

   def update(self, member_id: str, update_member_request: UpdateMemberRequest) -> Optional[Member]:
       if member_id not in self.members:
           return None

       current_member = self.members[member_id]
       update_data = update_member_request.model_dump(exclude_unset=True)

       for key, value in update_data.items():
           if value is not None:
               setattr(current_member, key, value)

       return current_member

   def delete(self, member_id: str) -> bool:
       if member_id not in self.members:
           return False

       del self.members[member_id]
       return True

 

實現 PostgreSQL 的儲存庫

# app/db/member_postgres_repository.py

from typing import List, Optional
from datetime import datetime
import uuid
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models.member import Member, MemberCreate, UpdateMemberRequest
from app.db.member_repository import MemberRepositoryInterface
from app.config import DATABASE_URL
from app.db.member_entity import MemberModel
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class PostgresMemberRepository(MemberRepositoryInterface):
   def get_all(self) -> List[Member]:
       with SessionLocal() as db:
           members = db.query(MemberModel).all()
           return [self._convert_to_member(member) for member in members]
   def get_by_id(self, member_id: str) -> Optional[Member]:
       with SessionLocal() as db:
           member = db.query(MemberModel).filter(MemberModel.id == member_id).first()
           if member:
               return self._convert_to_member(member)
           return None
   def create(self, member_create: MemberCreate) -> Member:
       member_id = str(uuid.uuid4())
       member_data = member_create.model_dump()
       db_member = MemberModel(
           id=member_id,
           first_name=member_data["first_name"],
           last_name=member_data["last_name"],
           age=member_data.get("age"),
           address=member_data.get("address"),
           birthday=member_data["birthday"],
           created_at=datetime.now(),
           created_by="system"
       )
       with SessionLocal() as db:
           db.add(db_member)
           db.commit()
           db.refresh(db_member)
       return self._convert_to_member(db_member)
   def update(self, member_id: str, update_member_request: UpdateMemberRequest) -> Optional[Member]:
       update_data = update_member_request.model_dump(exclude_unset=True)
       if not update_data:
           return self.get_by_id(member_id)
       with SessionLocal() as db:
           member = db.query(MemberModel).filter(MemberModel.id == member_id).first()
           if not member:
               return None
           for key, value in update_data.items():
               if value is not None:
                   setattr(member, key, value)
           db.commit()
           db.refresh(member)
           return self._convert_to_member(member)
   def delete(self, member_id: str) -> bool:
       with SessionLocal() as db:
           member = db.query(MemberModel).filter(MemberModel.id == member_id).first()
           if not member:
               return False
           db.delete(member)
           db.commit()
           return True
   def _convert_to_member(self, db_member: MemberModel) -> Member:
       """將 SQLAlchemy 模型轉換為 Pydantic 模型"""
       return Member(
           id=db_member.id,
           first_name=db_member.first_name,
           last_name=db_member.last_name,
           age=db_member.age,
           address=db_member.address,
           birthday=db_member.birthday,
           created_by=db_member.created_by,
           created_at=db_member.created_at
       )

 

DI Container 的實現與使用

使用 dependency-injector 實現依賴注入。

配置依賴注入容器

首先，我們需要設置一個配置檔，用於管理環境變數

# app/config.py

import os
from dotenv import load_dotenv

# 載入環境變數
load_dotenv()

# 資料庫連接設定
DATABASE_URL = os.getenv("DATABASE_URL")

def parse_bool(value, default=False):
   """將字符串轉換為布林值"""
   if value is None:
       return default
   value = value.lower()
   if value in ('true', 't', 'yes', 'y', '1'):
       return True
   elif value in ('false', 'f', 'no', 'n', '0'):
       return False
   else:
       return default

USE_POSTGRES = parse_bool(os.getenv("USE_POSTGRES", "false"))


 

然後，建立依賴注入容器

# app/di/container.py

from dependency_injector import containers, providers
from app.db.member_memory_repository import MemoryMemberRepository
from app.db.member_postgres_repository import PostgresMemberRepository
from app.config import USE_POSTGRES

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()
    # 環境變數配置
    # 如果設置了 USE_MEMORY，強制使用 MemoryMemberRepository
    # 否則根據 USE_POSTGRES 環境變數決定
    # 預設使用 memory repository
    # 當 USE_POSTGRES 為 true 時，才使用 postgres
    member_repository = providers.Singleton(
        PostgresMemberRepository
        if USE_POSTGRES
        else MemoryMemberRepository
    )

 

在 API 中使用依賴注入

在 FastAPI 中，我使用 `Depends` 來注入依賴

# app/api/members.py

from fastapi import APIRouter, HTTPException, status, Depends
from typing import List
from app.models.member import Member, MemberCreate, UpdateMemberRequest
from app.db.member_repository import MemberRepositoryInterface
from fastapi import Request

router = APIRouter(prefix="/members", tags=["members"])

# 從容器取得 member_repository
def get_member_repository(request: Request) -> MemberRepositoryInterface:
   return request.app.state.container.member_repository()

@router.get("", response_model=List[Member])
async def get_all_members(
   member_repository: MemberRepositoryInterface = Depends(get_member_repository)
):
   return member_repository.get_all()

@router.post("", response_model=Member, status_code=status.HTTP_201_CREATED)
async def create_member(
   member_create: MemberCreate,
   member_repository: MemberRepositoryInterface = Depends(get_member_repository)
):
   return member_repository.create(member_create)

# ... 其他 API 端點 ...

 

實作測試

測試是確保程式碼品質的重要環節。在我們的專案中，我們使用 pytest 來編寫和運行測試。

測試環境配置

首先，我們需要配置測試環境

# tests/conftest.py

import pytest
import sys
from pathlib import Path

# 將應用程序根目錄添加到 Python 路徑中
sys.path.append(str(Path(__file__).parent.parent))

 

測試資料庫的建立與清理

對於需要資料庫的測試，我需要在第一個測試執行的時候建立資料庫，最後一個測試結束後刪除資料庫

# tests/create_test_db.py

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
def create_test_database():
    """
    檢查 test_members_db 數據庫是否存在，如果不存在則創建
    """
    try:
        # 連接到 PostgreSQL 服務器
        conn = psycopg2.connect(
            host="localhost",
            user="postgres",
            password="postgres",
            port="5432"
        )
        conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
        cursor = conn.cursor()

        # 檢查數據庫是否存在
        cursor.execute("SELECT 1 FROM pg_database WHERE datname = 'test_members_db'")
        exists = cursor.fetchone()

        if not exists:
            print("Creating test_members_db database...")
            cursor.execute("CREATE DATABASE test_members_db")
            print("test_members_db database created successfully!")
        else:
            print("test_members_db database already exists.")

        cursor.close()
        conn.close()

    except Exception as e:
        print(f"Error: {e}")
        return False

    return True

def drop_test_database():
    """
    刪除 test_members_db 數據庫
    """
    try:
        # 連接到 PostgreSQL 服務器
        conn = psycopg2.connect(
            host="localhost",
            user="postgres",
            password="postgres",
            port="5432"
        )
        conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
        cursor = conn.cursor()

        # 檢查數據庫是否存在
        cursor.execute("SELECT 1 FROM pg_database WHERE datname = 'test_members_db'")
        exists = cursor.fetchone()

        if exists:
            print("Dropping test_members_db database...")
            # 確保沒有活動連接
            cursor.execute("""
                SELECT pg_terminate_backend(pg_stat_activity.pid)
                FROM pg_stat_activity
                WHERE pg_stat_activity.datname = 'test_members_db'
                AND pid <> pg_backend_pid()
            """)
            cursor.execute("DROP DATABASE test_members_db")
            print("test_members_db database dropped successfully!")
        else:
            print("test_members_db database does not exist.")

        cursor.close()
        conn.close()

    except Exception as e:
        print(f"Error dropping database: {e}")
        return False

    return True

 

API 測試的編寫

接下來，我們編寫 API 測試，使用 FastAPI 的 `TestClient`

# 導入被測目標物
from app.main import app

# 創建被測目標的客戶端
client = TestClient(app)

 

資料清理與驗證

在測試中，我們需要確保每個測試案例都是獨立的，這意味著我們需要在每個測試前後清理資料

@pytest.fixture(scope="session")
def test_initial():
    from tests.create_test_db import create_test_database, drop_test_database

    """創建並管理測試資料庫的生命週期"""
    print("\nSetting up test database...")
    create_test_database()

    yield

    # 測試結束後刪除整個測試資料庫
    print("\nCleaning up test database...")
    # drop_test_database()

@pytest.fixture(scope="module", autouse=True)
def setup_database(test_db_engine):
    """設置測試資料庫環境"""
    # 導入應用程序中定義的 Base
    from app.db.member_entity import Base,MemberModel

    # 創建所有表格
    Base.metadata.create_all(bind=test_db_engine)
    yield
    # 測試完成後刪除所有表格
    # Base.metadata.drop_all(bind=test_db_engine)


@pytest.fixture(autouse=True)
def clear_db(test_db_session_local):
    """在每次測試前清空會員數據庫"""
    # 導入被測目標物 - 在這裡導入以確保使用正確的環境變數
    from app.db.member_postgres_repository import MemberModel

    with test_db_session_local() as db:
        db.query(MemberModel).delete()
        db.commit()
    yield

 

對於 API 回傳結果，使用 assert 驗證

def test_create_member_postgres(self, test_db_session_local):
   """測試創建會員功能"""
   response = self.client.post("/api/v1/members", json=self.test_member_data)
   assert response.status_code == 201
   data = response.json()
   assert data["first_name"] == self.test_member_data["first_name"]
   # ... 其他斷言 ...

 

對於資料庫內容的驗證，查詢資料庫後再驗證

def test_create_member_postgres(self, test_db_session_local):
   """測試在 PostgreSQL 中創建會員"""
   # 創建會員
   response = self.client.post("/api/v1/members", json=self.test_member_data)
   assert response.status_code == 201
   member_id = response.json()["id"]
   
   # 直接從資料庫查詢驗證
   with SessionLocal() as db:
       db_member = db.query(MemberModel).filter(MemberModel.id == member_id).first()
       assert db_member is not None
       assert db_member.first_name == self.test_member_data["first_name"]
       assert db_member.last_name == self.test_member_data["last_name"]

 

執行所有測試

 

範例位置

https://github.com/yaochangyu/sample.dotblog/tree/master/Test/Lab.Py.ApiTest

希望這篇文章對你有所幫助！如果你有任何問題或建議，歡迎留言。

 

若有謬誤，煩請告知，新手發帖請多包涵


Microsoft MVP Award 2010~2017 C# 第四季
Microsoft MVP Award 2018~2022 .NET

Image result for microsoft+mvp+logo

系列文章

最新留言

本頁段落

標籤雲