FastAPI非ORM场景下测试数据库与开发数据库切换问题

我是一位非ORM用户，更倾向于使用原生SQL查询。在Free Code Camp学习FastAPI课程时，课程几乎围绕ORM展开，导致我在实现部分概念时遇到困难。

目前核心问题：编写接口测试时，需要让测试和CRUD操作指向测试数据库而非开发数据库。我正在使用testing.postgresql创建测试数据库，现有测试代码、数据库连接代码及配置代码如下：

现有测试代码 #

client = TestClient(app)

class TestUser:
    def setup_method(self, method):
        self.postgres = testing.postgresql.Postgresql(cache_initialized_db=True)
        self.db = psycopg2.connect(**self.postgres.dsn())

    def teardown_method(self, method):
        self.db.close()

    def test_get_users(self):
        res = client.get("/users")
        assert res.status_code == 200
        assert res.headers["content-type"] == "application/json"

现有数据库连接代码（未使用依赖） #

while True:
    try:
        conn = psycopg2.connect(
            database=settings.database,
            user=settings.user,
            port=settings.port,
            password=settings.password,
            cursor_factory=RealDictCursor,
        )
        cr = conn.cursor()
        console.print(
            "[green bold]SUCCESS[/]:    Connection To Database Established successfuly"
        )
        break
    except psycopg2.OperationalError:
        console.print(
            "[red bold]FAILED[/]:    Connection To Database Failed , Trying Again"
        )
        time.sleep(2)

现有配置代码 #

load_dotenv()

class Settings(BaseSettings):
    database: str
    user: str
    port: int
    password: str
    expiray: int
    algorithm: str
    secret_key: str

settings = Settings()

需要解决：如何让测试和CRUD操作指向测试数据库而非开发数据库？需要添加哪些关联代码？

1. 改用FastAPI依赖注入管理数据库连接 #

将硬编码的数据库连接改为FastAPI依赖，这样测试时可轻松替换连接对象：

创建dependencies.py文件（或直接写入主文件）：

from fastapi import Depends
import psycopg2
from psycopg2.extras import RealDictCursor
from .config import settings

def get_db_connection():
    conn = None
    try:
        conn = psycopg2.connect(
            database=settings.database,
            user=settings.user,
            port=settings.port,
            password=settings.password,
            cursor_factory=RealDictCursor,
        )
        yield conn
    finally:
        if conn:
            conn.close()

def get_db_cursor(conn=Depends(get_db_connection)):
    cur = conn.cursor()
    try:
        yield cur
    finally:
        cur.close()

在路由中使用该依赖（以/users接口为例）：

from fastapi import APIRouter, Depends
from psycopg2.extras import RealDictCursor
from .dependencies import get_db_cursor

router = APIRouter(prefix="/users")

@router.get("/")
def get_users(cur: RealDictCursor = Depends(get_db_cursor)):
    cur.execute("SELECT * FROM users;")
    users = cur.fetchall()
    return users

2. 调整配置，支持测试环境切换 #

修改Settings类，添加环境标识，方便测试时自动切换配置：

load_dotenv()

class Settings(BaseSettings):
    database: str
    user: str
    port: int
    password: str
    expiray: int
    algorithm: str
    secret_key: str
    # 默认开发环境，测试时可通过环境变量覆盖
    environment: str = "development"

    class Config:
        env_file = ".env"

settings = Settings()

3. 修改测试代码，替换数据库依赖 #

利用FastAPI的override_depends功能，让测试客户端使用测试数据库连接：

from fastapi.testclient import TestClient
from fastapi import Depends
import testing.postgresql
import psycopg2
from psycopg2.extras import RealDictCursor
from your_app_module import app, get_db_connection, get_db_cursor

client = TestClient(app)

class TestUser:
    def setup_method(self, method):
        # 启动测试PostgreSQL实例
        self.postgres = testing.postgresql.Postgresql(cache_initialized_db=True)
        # 创建测试数据库连接
        self.test_conn = psycopg2.connect(**self.postgres.dsn(), cursor_factory=RealDictCursor)
        
        # 初始化测试表结构（替换为你的实际表SQL）
        with self.test_conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE users (
                    id SERIAL PRIMARY KEY,
                    name VARCHAR(255) NOT NULL,
                    email VARCHAR(255) UNIQUE NOT NULL
                );
            """)
            self.test_conn.commit()
        
        # 覆盖依赖，返回测试连接
        def override_get_db_connection():
            yield self.test_conn
        
        def override_get_db_cursor(conn=Depends(override_get_db_connection)):
            cur = conn.cursor()
            try:
                yield cur
            finally:
                cur.close()
        
        app.dependency_overrides[get_db_connection] = override_get_db_connection
        app.dependency_overrides[get_db_cursor] = override_get_db_cursor

    def teardown_method(self, method):
        # 清理测试资源和依赖覆盖
        self.test_conn.close()
        self.postgres.stop()
        app.dependency_overrides.clear()

    def test_get_users(self):
        # 插入测试数据
        with self.test_conn.cursor() as cur:
            cur.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ("Test User", "test@example.com"))
            self.test_conn.commit()
        
        res = client.get("/users")
        assert res.status_code == 200
        assert res.headers["content-type"] == "application/json"
        assert len(res.json()) == 1
        assert res.json()[0]["name"] == "Test User"

4. （可选）优化测试数据库初始化 #

如果表结构复杂，可将建表SQL写入单独文件（如tests/schema.sql），测试时读取执行：

# 在setup_method中替换初始化表的代码
with open("tests/schema.sql", "r") as f:
    schema_sql = f.read()
with self.test_conn.cursor() as cur:
    cur.execute(schema_sql)
    self.test_conn.commit()

• 依赖注入是核心：通过依赖管理数据库连接，才能在测试时动态替换为测试库连接
• 测试数据库初始化：原生SQL需手动创建表结构，否则会出现表不存在的错误
• 依赖清理：测试结束后必须清除依赖覆盖，避免影响其他测试用例

内容的提问来源于stack exchange，提问作者kareem mahlees