FastAPI与Pydantic项目中如何Mock数据库？

我正在参与一个基于FastAPI和Pydantic的项目，任务来自GitHub招聘评估仓库，要求如下：

我们希望你以任何合适的方式模拟数据库交互层，并生成用于测试的数据（可以是随机数据）。该数据库层的实现由你决定。

这是我首次使用FastAPI，且从未接触过Mock技术。我已编写了如下代码：

from fastapi import FastAPI, Path
from typing import Optional
from pydantic import BaseModel, Field
import datetime as dt

class TradeDetails(BaseModel):
    buySellIndicator: str = Field(description="A value of BUY for buys, SELL for sells.")

    price: float = Field(description="The price of the Trade.")

    quantity: int = Field(description="The amount of units traded.")


class Trade(BaseModel):
    asset_class: Optional[str] = Field(alias="assetClass", default=None, description="The asset class of the instrument traded. E.g. Bond, Equity, FX...etc")

    counterparty: Optional[str] = Field(default=None, description="The counterparty the trade was executed with. May not always be available")

    instrument_id: str = Field(alias="instrumentId", description="The ISIN/ID of the instrument traded. E.g. TSLA, AAPL, AMZN...etc")

    instrument_name: str = Field(alias="instrumentName", description="The name of the instrument traded.")

    trade_date_time: dt.datetime = Field(alias="tradeDateTime", description="The date-time the Trade was executed")

    trade_details: TradeDetails = Field(alias="tradeDetails", description="The details of the trade, i.e. price, quantity")

    trade_id: str = Field(alias="tradeId", default=None, description="The unique ID of the trade")

    trader: str = Field(description="The name of the Trader")
     
app = FastAPI()

# data = {
#     'asset_class': 'Bond',
#     'conterparty': 'delio',
#     'instrument_id': 'AAPL',
#     'instrument_name': 'Guitar',
#     'trade_date_time':'2023-06-6 12:22',
#     'trade_details':{'buySellIndicator':'BUY', 'price':100.0, 'quantity': 10},
#     'trade_id': '11',
#     'trader':'john'
# }

trade = Trade(assetClass='asset',
    counterparty='count',
    instrumentId='AAPL',
    instrumentName='Guitar',
    tradeDateTime='2023-06-6 12:22',
    tradeDetails={'buySellIndicator':'BUY', 'price':100.0, 'quantity': 10},
    tradeId='11',
    trader='john')

@app.get("/")
def index():
    return {"name": "API Developer Assessment"}

@app.get("/get-trade/{tradeId}")
def get_trade(tradeId:int=Path(description="The Id of the trade you want to view")):
    return trade.tradeId

@app.get("/Trade")
def get_trade_list():
    return trade.dict()

# @app.get('/get-by-query')
# def get_trade()

实现Mock数据库的实用方案 #

因为是测试场景，不需要真实数据库，用内存结构+随机数据生成就能快速搞定，完全符合需求。

1. 用内存字典模拟数据库表 #

直接用Python字典存储交易数据，键为trade_id，值为Trade对象，查询、新增操作都很高效：

# 全局变量，模拟数据库的交易表
mock_trade_db = {}

2. 生成随机测试数据 #

用faker库快速生成真实感的随机数据，先安装依赖：

pip install faker

然后写一个生成随机Trade的函数：

from faker import Faker
import random
from datetime import datetime

fake = Faker()

def generate_random_trade(trade_id: str = None) -> Trade:
    # 定义可选的字段值范围
    asset_classes = ["Bond", "Equity", "FX", "Commodity", None]
    buy_sell_options = ["BUY", "SELL"]
    
    return Trade(
        assetClass=random.choice(asset_classes),
        # 20%概率不提供交易对手
        counterparty=fake.company() if random.random() > 0.2 else None,
        instrumentId=fake.lexify(text="????"),  # 生成4位随机代码
        instrumentName=fake.catch_phrase(),
        tradeDateTime=fake.date_time_this_year(),
        tradeDetails={
            "buySellIndicator": random.choice(buy_sell_options),
            "price": round(random.uniform(10.0, 1000.0), 2),
            "quantity": random.randint(1, 100)
        },
        tradeId=trade_id or fake.uuid4(),  # 自动生成唯一ID如果没提供
        trader=fake.name()
    )

3. 初始化Mock数据库 #

启动API时预生成几条测试数据，避免空库：

# 启动时生成5条随机交易数据
for i in range(5):
    random_trade = generate_random_trade(f"trade_{i+1}")
    mock_trade_db[random_trade.trade_id] = random_trade

4. 修改接口适配Mock数据库 #

把原来的接口改成从mock_trade_db读取/写入数据，同时处理不存在的情况：

@app.get("/get-trade/{tradeId}")
def get_trade(tradeId: str = Path(description="The Id of the trade you want to view")):
    # 从Mock数据库查询交易
    target_trade = mock_trade_db.get(tradeId)
    if not target_trade:
        return {"error": "Trade not found"}
    # 用别名返回数据，符合API字段命名规范
    return target_trade.dict(by_alias=True)

@app.get("/Trade")
def get_trade_list():
    # 返回所有交易数据
    return [trade.dict(by_alias=True) for trade in mock_trade_db.values()]

# 可选：新增创建交易的接口，完善CRUD功能
@app.post("/Trade")
def create_trade(trade: Trade):
    # 如果没有提供trade_id，自动生成
    if not trade.trade_id:
        trade.trade_id = fake.uuid4()
    mock_trade_db[trade.trade_id] = trade
    return {"message": "Trade created successfully", "trade_id": trade.trade_id}

5. 进阶：用类封装Mock数据库（可选） #

如果想让结构更清晰，模拟真实数据库的CRUD方法，可以封装成类：

class MockTradeDB:
    def __init__(self):
        self._db = {}
    
    def get_by_id(self, trade_id: str) -> Optional[Trade]:
        return self._db.get(trade_id)
    
    def get_all(self) -> list[Trade]:
        return list(self._db.values())
    
    def add(self, trade: Trade):
        self._db[trade.trade_id] = trade
    
    def delete(self, trade_id: str) -> bool:
        if trade_id in self._db:
            del self._db[trade_id]
            return True
        return False

# 实例化Mock数据库
mock_db = MockTradeDB()

# 预生成测试数据
for i in range(5):
    mock_db.add(generate_random_trade(f"trade_{i+1}"))

# 接口中使用封装后的Mock数据库
@app.get("/get-trade/{tradeId}")
def get_trade(tradeId: str = Path(description="The Id of the trade you want to view")):
    trade = mock_db.get_by_id(tradeId)
    if not trade:
        return {"error": "Trade not found"}
    return trade.dict(by_alias=True)

内容的提问来源于stack exchange，提问作者Karthik Bhandary